使用RabbitMQ等消息代理时，应由Controller还是Service负责发送消息？

这两种方案都不是最优解，业界通用最佳实践是引入领域事件+独立事件发布器的分层设计，同时规避两者的痛点。

原有方案的核心缺陷 #

Controller层发布消息 #

• 违反DRY原则：多个Controller调用同个Service方法时，需要重复编写消息发布逻辑，漏写就会导致事件丢失
• 职责越界：Controller层本应只负责HTTP请求参数校验、响应格式封装，耦合MQ发布逻辑后会导致单文件职责过重

Service层发布消息 #

• 方法签名污染：同个Service方法被内部RPC、定时任务调用时往往不需要发消息，新增标记参数会导致方法越来越臃肿
• 耦合中间件：Service层直接依赖MQ实现，后续更换消息中间件时需要改动所有相关业务代码，维护成本极高

最优解：领域事件发布订阅方案 #

核心逻辑是拆分职责：Service只负责触发业务事件，不关心事件的投递逻辑；独立的订阅组件统一处理MQ消息发送，各层完全解耦。

代码实现 #

// 1. 通用事件发布器（全局单例）
class EventPublisher {
  private static instance: EventPublisher;
  private subscribers = [];

  static getInstance() {
    if (!EventPublisher.instance) EventPublisher.instance = new EventPublisher();
    return EventPublisher.instance;
  }

  publish(event) {
    this.subscribers.forEach(sub => sub(event));
  }

  subscribe(eventType, handler) {
    this.subscribers.push(event => {
      if (event.type === eventType) handler(event.data);
    });
  }
}

// 2. Service层仅发布领域事件，完全不感知MQ存在
class ExampleService {
  async thing() {
    // 核心业务逻辑
    const result = { /* 业务返回数据 */ };
    // 触发业务事件，不需要关心后续投递逻辑
    EventPublisher.getInstance().publish({
      type: 'thing_completed',
      data: result
    });
    return result;
  }
}

// 3. 独立MQ事件订阅组件，统一处理消息发送
class RabbitMQEventSubscriber {
  constructor(rabbitMQ) {
    // 订阅业务事件，触发后自动发送MQ消息
    EventPublisher.getInstance().subscribe('thing_completed', async (data) => {
      await rabbitMQ.sendMessage('thing_exchange', 'thing.routing.key', data);
    });
  }
}

// 4. Controller层不需要做任何修改，也不需要感知MQ
class ExampleController {
  constructor(exampleService) {
    this.exampleService = exampleService;
  }

  async index() {
    return await this.exampleService.thing();
  }
}

方案优势 #

• 无重复代码：消息发布逻辑统一在订阅组件中，不需要每个Controller调用Service后重复编写
• 无方法污染：Service不需要新增任何标记参数，内部调用、定时任务调用时只要不注册对应事件的MQ订阅就不会发送消息
• 易扩展：新增事件仅需要加对应事件的发布和订阅逻辑，更换MQ仅需要修改订阅组件的实现，不需要改动业务代码
• 完全符合单一职责原则，各层耦合度极低

小项目简化方案 #

如果项目规模较小、后续扩展需求不多，可以用装饰器/切面的方式简化实现，不需要单独抽事件发布器：

// 消息发布装饰器
function PublishMQEvent(eventType) {
  return function (target, propertyKey, descriptor) {
    const originalMethod = descriptor.value;
    descriptor.value = async function (...args) {
      const result = await originalMethod.apply(this, args);
      await this.rabbitMQ.sendMessage(eventType, result);
      return result;
    };
    return descriptor;
  };
}

class ExampleService {
  constructor(rabbitMQ) {
    this.rabbitMQ = rabbitMQ;
  }

  // 需要发消息的方法加装饰器即可，不需要修改方法内部逻辑
  @PublishMQEvent('thing_completed')
  async thing() {
    // 核心业务逻辑
    return { /* 业务数据 */ };
  }
}

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