在微服务架构中，服务之间经常通过AMQP协议（比如RabbitMQ）进行异步通信。但当你编写单元测试或者本地联调时，如果每次都要启动一个真实的RabbitMQ实例，测试会变得又慢又重，而且在CI环境中维护消息中间件也是一件麻烦事。更合理的做法是自己实现一个AMQP Mock服务，让被测代码以为自己连上了真实的消息中间件。本文将围绕Node.js环境，详细讲解AMQP Mock的实现思路和完整代码。
AMQP协议的核心交互流程
要实现Mock，首先要明白真实的AMQP 0-9-1协议是怎么工作的。客户端与服务器建立TCP连接后，会经历一个握手过程：客户端发送protocol header，服务端返回connection.start，客户端回connection.start-ok，随后双方交换tune、open等帧，连接才算建立完成。之后客户端通过channel.open创建通道，在通道上执行exchange.declare、queue.declare、queue.bind等指令，最后用basic.publish发送消息、basic.consume订阅消息。
这套流程是二进制帧协议，帧头8字节包含type、channel、size等字段。如果要从零实现协议解析，工作量不小。所以在动手之前，建议先评估你的测试目标：如果只是验证业务代码的收发逻辑，用高层封装的Mock就够了；如果需要验证协议层行为（比如channel错误处理），才需要实现真正的帧解析。下面介绍两种不同深度的方案。
方案一：基于amqplib的内存级Mock
最简单的思路是引入一个进程内的Mock模块，替换真实的amqplib连接。被测代码通常这样获取连接：
const amqp = require('amqplib');
async function getConnection() {
return amqp.connect('amqp://localhost');
}我们可以在测试启动时，通过依赖注入或者模块替换的方式，把amqplib替换成一个内存实现。核心是模拟connect返回的对象链：connection上有createChannel方法，channel上有assertQueue、sendToQueue、consume等。下面是一个最小可用的Mock实现：
class MockChannel {
constructor() {
this.queues = new Map(); // 队列名 -> 消息数组
this.consumers = new Map(); // 队列名 -> 回调
}
async assertQueue(queue) {
if (!this.queues.has(queue)) {
this.queues.set(queue, []);
}
return { queue, messageCount: this.queues.get(queue).length };
}
async sendToQueue(queue, content) {
if (!this.queues.has(queue)) {
this.queues.set(queue, []);
}
this.queues.get(queue).push(content);
const cb = this.consumers.get(queue);
if (cb) {
cb({ content });
}
return true;
}
async consume(queue, onMessage) {
this.consumers.set(queue, onMessage);
// 把历史积压消息推给消费者
const pending = this.queues.get(queue) || [];
while (pending.length) {
onMessage({ content: pending.shift() });
}
return { consumerTag: 'mock-tag-' + Date.now() };
}
async ack() {}
async nack() {}
async close() {}
}
class MockConnection {
async createChannel() {
return new MockChannel();
}
async close() {}
}
const mockAmqp = {
async connect() {
return new MockConnection();
}
};
module.exports = mockAmqp;在测试文件中，用proxyquire或者简单的模块缓存替换，就能让被测代码用上这个Mock：
const proxyquire = require('proxyquire');
const mod = proxyquire('./producer', {
'amqplib': mockAmqp
});
// 此时mod内部拿到的connect已经是内存版实现这种方案的优点是实现简单、运行速度快，完全不涉及网络IO，整个测试套件跑下来毫秒级完成。缺点是它只覆盖了amqplib暴露的API行为，无法模拟网络中断、心跳超时等异常场景，而且Mock行为和真实服务端语义可能有偏差，比如RabbitMQ在消息不可路由时的处理逻辑，这个内存Mock是模拟不出来的。
方案二：基于net模块实现真实的AMQP Mock服务器
如果希望被测代码不做任何改动、真实地走一遍TCP连接，那就需要用Node.js的net模块起一个本地服务，在指定端口上响应AMQP协议帧。这要求至少实现握手阶段的帧交互。AMQP帧的基本结构是：1字节type、2字节channel、4字节payload长度、payload内容、1字节帧结束符0xCE。握手时客户端先发送8字节的协议头"AMQP\0\0\x09\x01"，服务端需要回connection.start帧。
下面是一个简化版的握手Mock服务器示例：
const net = require('net');
const FRAME_END = 0xCE;
// 编码一个AMQP帧
function buildFrame(type, channel, payload) {
const header = Buffer.alloc(8);
header.writeUInt8(type, 0);
header.writeUInt16BE(channel, 1);
header.writeUInt32BE(payload.length, 3);
header.writeUInt8(FRAME_END, 7);
// 实际应把结束符放在payload之后，这里简化处理
return Buffer.concat([header.slice(0, 7), payload, Buffer.from([FRAME_END])]);
}
const server = net.createServer(socket => {
socket.on('data', buf => {
// 客户端发的protocol header
if (buf.slice(0, 4).toString() === 'AMQP') {
// 回复connection.start，payload为简化后的方法帧
const payload = Buffer.from([
0x00, 0x0a, 0x00, 0x0a // class-id=10, method-id=10 (connection.start)
]);
socket.write(buildFrame(1, 0, payload));
}
});
socket.on('error', err => console.log('socket error:', err.message));
});
server.listen(5673, () => {
console.log('AMQP mock server listening on 5673');
});完整实现协议帧的方法编码比较繁琐，因为每个方法帧里的字段（如shortstr、longstr、bits）都要按规范编码解码。建议直接参考amqplib源码里的codegen模块，它内部有根据RabbitMQ规范XML生成的编解码代码，可以复用。这种方案的最大价值在于可以做真实网络层测试，比如主动断开连接来验证客户端的重连逻辑：
// 模拟服务端异常断连，测试客户端重连
function killAllClients() {
for (const sock of server.connections) {
sock.destroy(); // 直接销毁socket
}
}代价是实现和维护成本高。一个折中的做法是使用社区现成的工具，比如在Docker里跑一个轻量的RabbitMQ容器做集成测试，而单元测试仍用方案一的内存Mock，两者结合兼顾速度和真实性。
进阶技巧：断言投递与异常模拟
光能收发消息还不够，测试中经常需要断言消息确实被发送到了正确的队列。可以在MockChannel上增加记录能力：
// 在MockChannel中增加记录
constructor() {
this.published = []; // 记录所有发出去的消息
}
async sendToQueue(queue, content, options) {
this.published.push({ queue, content, options, time: Date.now() });
// ...原有逻辑
}
// 测试断言
const chai = require('chai');
const expect = chai.expect;
it('订单创建后应发送消息到order队列', async () => {
await createOrder({ id: 1 });
const msg = channel.published.find(m => m.queue === 'order');
expect(msg).to.exist;
expect(JSON.parse(msg.content.toString()).orderId).to.equal(1);
});异常模拟也是高价值场景。可以给Mock加上可配置的故障开关，模拟消息回滚、队列满、连接超时等情况：
class FlakyChannel extends MockChannel {
constructor() {
super();
this.failMode = null; // 'reject' | 'timeout' | null
}
async sendToQueue(queue, content) {
if (this.failMode === 'timeout') {
await new Promise(r => setTimeout(r, 5000));
}
if (this.failMode === 'reject') {
throw new Error('NO_ROUTE: 消息不可路由');
}
return super.sendToQueue(queue, content);
}
}这样就能覆盖业务代码里的错误处理分支，比如发送失败后写补偿表、触发告警等逻辑。另外注意一点：如果被测代码使用了confirm模式（等待服务端basic.ack确认），Mock也要相应返回confirm帧，否则代码会一直等待。这类细节建议对照amqplib的API文档逐一核对，把Mock的行为对齐真实语义，测试结果才有参考价值。
总结一下，AMQP Mock的实现深度取决于测试需求：验证业务逻辑用内存Mock足够，验证网络与协议行为则需要net级别的模拟服务器。无论哪种方案，都建议把Mock实现和断言工具封装成独立的测试辅助模块，在多个项目间复用，长期来看能显著降低消息队列相关测试的维护成本。