如何在NestJS中通过ClientProxy.emit发送RabbitMQ自定义负载？

问题背景 #

我正在将NestJS微服务集成到某系统中，该系统要求生成格式为{"channelId":"1ed95c30-a5ad-64ec-9aef-bbad313f5c4c"}的RabbitMQ消息负载。但NestJS的@nestjs/microservices.ClientProxy.emit方法的负载格式化规则与需求不兼容：该方法接收pattern和data两个参数，会将它们作为顶层属性生成负载，例如调用emit('', '1ed95c30-a5ad-64ec-9aef-bbad313f5c4c')会生成{"pattern":"","data":{"channelId":"1ed95c30-a5ad-64ec-9aef-bbad313f5c4c"}}。想知道是否有办法发送自定义负载，还是不应使用@nestjs/microservices？

解决方案 #

1. 自定义消息序列化器 #

NestJS允许通过自定义序列化器覆盖默认消息格式，实现Serializer接口即可自定义序列化逻辑：

import { Serializer } from '@nestjs/microservices';

export class CustomRabbitSerializer implements Serializer {
  serialize(value: any): any {
    // 直接返回传入的data，跳过默认的pattern和data包装结构
    return value.data;
  }
}

创建ClientProxy时配置这个序列化器：

import { ClientProxyFactory, Transport } from '@nestjs/microservices';

const rabbitClient = ClientProxyFactory.create({
  transport: Transport.RMQ,
  options: {
    urls: ['amqp://localhost:5672'],
    queue: 'target_queue',
    serializer: new CustomRabbitSerializer(),
  },
});

配置完成后，调用emit时传入的data会直接作为消息负载发送，不再被pattern和data包裹。

2. 直接使用原生AMQP客户端 #

如果自定义序列化器仍无法满足需求，或者需要完全控制消息发送流程，可以直接使用amqplib这类原生AMQP客户端，绕过NestJS的ClientProxy封装：

import * as amqp from 'amqplib';

async function sendCustomPayload() {
  const connection = await amqp.connect('amqp://localhost:5672');
  const channel = await connection.createChannel();
  const targetQueue = 'target_queue';
  
  await channel.assertQueue(targetQueue, { durable: false });
  const payload = JSON.stringify({ channelId: '1ed95c30-a5ad-64ec-9aef-bbad313f5c4c' });
  
  channel.sendToQueue(targetQueue, Buffer.from(payload));
  
  setTimeout(() => {
    connection.close();
  }, 500);
}

这种方式完全脱离NestJS微服务的封装逻辑，能实现任意格式的消息发送。

3. 调整emit调用参数配合序列化器 #

如果场景允许，可直接将完整的自定义负载作为data参数传入，再结合自定义序列化器移除包装：

rabbitClient.emit('', { channelId: '1ed95c30-a5ad-64ec-9aef-bbad313f5c4c' });

搭配前面的CustomRabbitSerializer，即可直接发送符合要求的负载格式。

结论 #

完全可以继续使用@nestjs/microservices，自定义序列化器就能解决负载格式不兼容的问题。如果需要更极致的控制，再考虑使用原生AMQP客户端。

内容的提问来源于stack exchange，提问作者Coder1