Spring Cloud Bus 消息总线源码阅读

Spring Cloud Bus 把微服务节点连成一条"总线"：任意节点发事件，其他节点通过消息中间件（默认 RabbitMQ/Kafka）接收。最常见的场景是配置中心变更后广播刷新。本文从源码拆解总线的完整链路。

Bus 架构全景

服务节点 A ──▶ Bus（消息中间件：RabbitMQ/Kafka）
                   │
                   ├──▶ 服务节点 B（收到远程事件）
                   ├──▶ 服务节点 C（收到远程事件）
                   └──▶ 服务节点 D（收到远程事件）

核心组件：
  BusAutoConfiguration        自动配置
  SpringCloudBusConfiguration 基础设施
  BusBridge                   事件桥接（本地事件 ↔ 远程事件）
  @RemoteApplicationEventScan 远程事件扫描
  RemoteApplicationEvent      远程事件基类

消息通道：SpringCloudBusInput / Output

通道定义

Bus 用 Spring Integration 的通道收发消息：

java

// spring-cloud-bus 的绑定接口
public interface SpringCloudBusInput {
    @Input(SpringCloudBusClient.INPUT)
    SubscribableChannel input();    // 接收远程事件
}

public interface SpringCloudBusOutput {
    @Output(SpringCloudBusClient.OUTPUT)
    MessageChannel output();        // 发送远程事件
}

发事件：service.publish() → BusAutoConfiguration
                → output 通道 → MQ（bus.topic 主题）
收事件：MQ（bus.topic）→ input 通道 → 事件反序列化 → 本地分发

主题

RabbitMQ：SpringCloudBus（topic 交换机）
Kafka：   springCloudBus（主题）

所有节点的 bus 消息共用同一主题，事件通过 destination 字段路由（可指定目标服务）。

RemoteApplicationEvent：远程事件基类

类结构

java

// org.springframework.cloud.bus.event.RemoteApplicationEvent
public abstract class RemoteApplicationEvent extends ApplicationEvent {

    private static final long serialVersionUID = 1L;

    private final String originService;     // 事件来源服务实例
    private final String destinationService;// 目标服务（* 表示广播）
    private final String id;                // 事件 ID（幂等）

    protected RemoteApplicationEvent(Object source, String originService,
                                     String destinationService) {
        super(source);
        this.originService = originService;
        this.destinationService = destinationService;
        this.id = UUID.randomUUID().toString();
    }

    // 不序列化 source（JVM 内部对象），用数据替代
    @JsonIgnore
    @Override
    public Object getSource() {
        return super.getSource();
    }
}

事件序列化策略

java

// 反序列化时用 source（Object）还原：
// Jackson 反序列化 RemoteApplicationEvent 时，
// source 字段被解析为事件内部数据（如配置变更内容）

常见远程事件

事件	作用
RefreshRemoteApplicationEvent	触发 `/actuator/refresh` 配置刷新
EnvironmentChangeRemoteApplicationEvent	通知环境变量变更
AckRemoteApplicationEvent	确认收到（回执）
SentRemoteApplicationEvent	发布确认

@RemoteApplicationEventScan：远程事件扫描

作用

Bus 需要知道有哪些自定义远程事件类型，用于反序列化时识别。

java

// 启动类上标注
@SpringBootApplication
@RemoteApplicationEventScan(basePackages = "com.example.events")
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

源码

java

// org.springframework.cloud.bus.event.RemoteApplicationEventScan
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Documented
@Import(RemoteApplicationEventScanRegistrar.class)
public @interface RemoteApplicationEventScan {
    String[] value() default {};
    String[] basePackages() default {};   // 扫描包
    Class<?>[] basePackageClasses() default {};
}

java

// RemoteApplicationEventScanRegistrar
public class RemoteApplicationEventScanRegistrar implements ImportBeanDefinitionRegistrar {

    @Override
    public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry registry) {
        // 1. 扫描指定包下所有 RemoteApplicationEvent 子类
        // 2. 收集事件类型列表
        // 3. 注册 SubtypeElementConverter（Jackson 类型映射）
    }
}

为什么需要扫描

Bus 反序列化远程事件时，JSON 里只有类型名（@class 或 type 字段），Jackson 需要知道"这个类型是什么"，扫描注册后才能在反序列化时还原为具体事件类：

json

{
  "@class": "com.example.events.OrderCreatedRemoteEvent",
  "originService": "order-service:8080",
  "destinationService": "*"
}

BusBridge：事件桥接

职责

本地 Spring 事件 → 发到总线（远程）
总线消息（远程） → 转成本地 Spring 事件

java

// org.springframework.cloud.bus.BusBridge
public interface BusBridge {
    void send(String id, String originService, String destinationService,
              String event, String eventType);   // 发送远程事件
}

java

// 默认实现：BusAutoConfiguration 中的 MessageChannelBusBridge
public class MessageChannelBusBridge implements BusBridge {

    private final MessageChannel output;   // bus 输出通道

    @Override
    public void send(String id, String originService, String destinationService,
                     String event, String eventType) {
        // 构建消息头（类型/来源/目标/ID）
        Map<String, Object> headers = new HashMap<>();
        headers.put("id", id);
        headers.put("originService", originService);
        headers.put("destinationService", destinationService);
        headers.put("eventType", eventType);

        // 发送到 MQ
        output.send(MessageBuilder.withPayload(event).copyHeaders(headers).build());
    }
}

发送与接收的完整链路

发送侧

@RefreshScope 刷新 / 配置变更
   │
   ▼
RefreshRemoteApplicationEvent 发布到本地 Spring 容器
   │
   ▼
ApplicationListener（EventListener）捕获
   │
   ▼
BusAutoConfiguration 中：
   ServiceMatcher + ApplicationEventMulticaster
   │
   ▼
MessageChannelBusBridge.send() → bus 输出通道
   │
   ▼
Spring Integration → MQ（RabbitMQ/Kafka）→ 所有节点

接收侧

MQ → SpringCloudBusInput.input() 通道
   │
   ▼
（Spring Integration 消息）
   │
   ▼
BusAutoConfiguration 的 Listener：
   ObjectMapper 反序列化（用扫描注册的类型）
   │
   ▼
ServiceMatcher 校验目标（destination 匹配自己？）
   │
   ▼
还原为 RemoteApplicationEvent → 发布到本地 Spring 容器
   │
   ▼
本地监听器处理（如 Refresh 事件的 ApplicationListener）

核心源码片段

java

// BusAutoConfiguration 接收消息
@EventListener(classes = RemoteApplicationEvent.class)
public void acceptLocal(RemoteApplicationEvent event) {
    // 本地事件 → 转发到总线
    if (this.serviceMatcher.isFromSelf(event)) {
        this.busBridge.send(...);
    }
}

// 接收远程消息
@StreamListener(SpringCloudBusClient.INPUT)
public void acceptRemote(RemoteApplicationEvent event) {
    // 目标服务校验
    if (this.serviceMatcher.isForSelf(event)) {
        // 还原并发布到本地
        applicationEventPublisher.publishEvent(event);
    }
}

ServiceMatcher：服务匹配

作用

避免事件无限循环（自己发的事件又收回来）：

java

// org.springframework.cloud.bus.ServiceMatcher
public class ServiceMatcher {

    private final String serviceId;   // 当前服务实例 ID

    // 是否自己发的
    public boolean isFromSelf(RemoteApplicationEvent event) {
        return event.getOriginService().equals(serviceId);
    }

    // 是否发给自己
    public boolean isForSelf(RemoteApplicationEvent event) {
        String destinationService = event.getDestinationService();
        return "*".equals(destinationService)   // 广播
                || destinationService.equals(serviceId);
    }
}

自定义事件总线

自定义远程事件

java

// 1. 定义事件（继承 RemoteApplicationEvent）
public class OrderCreatedRemoteEvent extends RemoteApplicationEvent {
    private Long orderId;
    private String orderNo;

    // 必须有默认构造（反序列化用）
    public OrderCreatedRemoteEvent() {
        super(new Object(), "", "*");
    }

    public OrderCreatedRemoteEvent(Object source, String originService,
                                   Long orderId, String orderNo) {
        super(source, originService, "*");
        this.orderId = orderId;
        this.orderNo = orderNo;
    }
    // getter/setter
}

java

// 2. 扫描注册
@SpringBootApplication
@RemoteApplicationEventScan(basePackages = "com.example.events")
public class Application { ... }

java

// 3. 发送
@Service
public class EventPublisher {
    @Autowired
    private ApplicationEventPublisher publisher;

    @Autowired
    private ServiceMatcher serviceMatcher;

    public void publishOrderCreated(Long orderId, String orderNo) {
        publisher.publishEvent(new OrderCreatedRemoteEvent(
                this, serviceMatcher.getServiceId(), orderId, orderNo));
    }
}

java

// 4. 接收（任意节点）
@Component
public class OrderEventListener {
    @EventListener
    public void onOrderCreated(OrderCreatedRemoteEvent event) {
        // 所有节点都会收到（destination = *）
        log.info("收到订单创建事件: {}", event.getOrderNo());
    }
}

定向发送

java

// destination 指定具体服务，只有该服务收到
public class OrderCreatedRemoteEvent extends RemoteApplicationEvent {
    public OrderCreatedRemoteEvent(Object source, String originService,
                                   String destinationService, ...) {
        super(source, originService, destinationService);
    }
}

常见问题

Bus 依赖什么中间件？ 默认 RabbitMQ/Kafka，通过 spring-cloud-bus 的绑定器（Binder）对接。
事件收不到？ 检查总线主题是否一致、destination 是否匹配、@RemoteApplicationEventScan 是否扫描到事件类。
事件循环？ ServiceMatcher 会过滤自己发的事件（isFromSelf），不要覆盖该逻辑。
自定义事件反序列化失败？ 事件类需要无参构造 + getter/setter + 默认构造，且被扫描注册。

Spring Cloud Bus 消息总线源码阅读 ​

Bus 架构全景 ​

消息通道：SpringCloudBusInput / Output ​

通道定义 ​

主题 ​

RemoteApplicationEvent：远程事件基类 ​

类结构 ​

事件序列化策略 ​

常见远程事件 ​

@RemoteApplicationEventScan：远程事件扫描 ​

作用 ​

源码 ​

为什么需要扫描 ​

BusBridge：事件桥接 ​

职责 ​

发送与接收的完整链路 ​

发送侧 ​

接收侧 ​

核心源码片段 ​

ServiceMatcher：服务匹配 ​

作用 ​

自定义事件总线 ​

自定义远程事件 ​

定向发送 ​

常见问题 ​

Spring Cloud Bus 消息总线源码阅读

Bus 架构全景

消息通道：SpringCloudBusInput / Output

通道定义

主题

RemoteApplicationEvent：远程事件基类

类结构

事件序列化策略

常见远程事件

@RemoteApplicationEventScan：远程事件扫描

作用

源码

为什么需要扫描

BusBridge：事件桥接

职责

发送与接收的完整链路

发送侧

接收侧

核心源码片段

ServiceMatcher：服务匹配

作用

自定义事件总线

自定义远程事件

定向发送

常见问题