跳到主要内容

消息与事件

Coco 消息（coco-messaging）提供进程内的发布/订阅边界：业务用 CocoMessagePublisher 发消息，用 @CocoMessageListener 声明订阅，投递方式、失败处理和关闭行为都由配置决定。

它绑定 coco.messaging 命名空间，默认关闭，需显式打开 coco.messaging.enabled=true。

明确的能力边界

这个模块是本地事件总线加一层可替换的传输 SPI，不是 MQ 集成。持久化、跨进程投递、重试、死信队列、事务消息和 exactly-once 都不在范围内 —— 需要这些能力时，实现 CocoMessageTransport 接入 RabbitMQ / Kafka / RocketMQ，业务代码不用改。

功能简介​

  • 发布/订阅：CocoMessagePublisher.publish(topic, payload) 发布，@CocoMessageListener(topic = "...") 订阅，支持 order 控制同一主题内的处理顺序。
  • 同步或异步投递：delivery-mode 决定在调用线程处理还是交给有界队列异步处理。
  • 失败策略：订阅者抛异常时，可选择快速失败或记录后继续。
  • 无订阅者策略：发到没人订阅的主题时，可选择忽略或报错——后者能及早暴露主题拼写错误。
  • 关闭策略：应用关闭时，异步队列里的消息可选择排空或丢弃。
  • 可替换传输：CocoMessageTransport 是 SPI，默认 LocalCocoMessageTransport 仅进程内有效；提供自定义 Bean 即可换成外部消息中间件。

如何启用接入​

1. 打开开关​

coco:
  messaging:
    enabled: true
    delivery-mode: sync              # sync | async，默认 sync
    failure-policy: fail-fast        # fail-fast | log-and-continue
    no-subscriber-policy: ignore     # ignore | fail
    async:
      queue-capacity: 1024
      shutdown-await: 5s
      shutdown-policy: drain         # drain | discard

2. 发布消息​

@Service
class OrderService {

    private final CocoMessagePublisher publisher;

    OrderService(CocoMessagePublisher publisher) {
        this.publisher = publisher;
    }

    void create(CreateOrderRequest request) {
        Order order = this.orderRepository.save(request.toOrder());
        this.publisher.publish("order.created", order.id());
    }
}

3. 订阅消息​

@Component
class OrderCreatedListeners {

    @CocoMessageListener(topic = "order.created")
    void sendConfirmation(Long orderId) {
        this.notificationService.notifyCustomer(orderId);
    }

    @CocoMessageListener(topic = "order.created", order = 10)
    void updateStatistics(Long orderId) {
        this.statisticsService.increment(orderId);
    }
}

order 小的先执行。同步模式下，监听方法在发布者线程内运行，因此会落在发布方的事务边界内。

关键配置项​

前缀 coco.messaging。

配置项类型默认值说明
enabledbooleanfalse是否启用消息模块
delivery-modeenumsyncsync 在调用线程处理；async 投入有界队列
failure-policyenumfail-fastfail-fast 向发布者抛出；log-and-continue 记录后继续通知其余订阅者
no-subscriber-policyenumignore主题无订阅者时忽略或报错
async.queue-capacityint1024异步队列容量
async.shutdown-awaitDuration5s关闭时等待队列处理的时长
async.shutdown-policyenumdrain超时后排空剩余消息或直接丢弃

投递模式的取舍​

syncasync
执行线程发布者线程队列工作线程
事务落在发布者事务内脱离发布者事务
发布者感知失败能（fail-fast 时抛出）不能
队列满时不涉及按背压策略处理

同步模式下监听器在发布者的事务里执行，意味着监听器失败会回滚业务事务 —— 这既可能是你想要的（强一致），也可能不是（通知失败不该回滚订单）。需要后者时用异步模式，或在监听器内自行捕获异常。

接入外部消息中间件​

实现 CocoMessageTransport 并注册为 Bean，即可替换默认的进程内传输，业务侧的发布和订阅代码不用改：

@Bean
public CocoMessageTransport cocoMessageTransport(RabbitTemplate template) {
    return new RabbitCocoMessageTransport(template);
}

SPI 只描述发布与订阅两个动作。持久化、重试、死信队列、顺序保证和 exactly-once 由你的实现决定 —— 框架不假装提供这些保证。

边界注意事项​

  • 仅进程内：默认实现的订阅关系只在当前 JVM 有效，多实例部署下各实例互不感知。跨实例投递需要自定义传输实现。
  • 不持久化：进程重启后队列内容丢失。需要投递保证时不要依赖默认实现。
  • 异步模式脱离事务：异步投递发生在发布者事务提交之后没有保证，因此不能用它实现"事务内可靠投递"。需要最终一致性时，配合本地消息表等模式，由你的传输实现负责。
  • 无内建重试：failure-policy 只决定失败后是否继续通知其余订阅者，不会重投。