Spring Cloud Bus的消息发布与订阅

2021-04-28 1,073 0

介绍

Spring Cloud Bus 对自己的定位是 Spring Cloud 体系内的消息总线，他是基于Spring Cloud Stream进一步的封装，结合Spring Context自身事件机制，来快速简单地完成分布式集群中服务的业务事件定义、业务消息的发送与订阅流程

https://docs.spring.io/spring-cloud-bus/docs/current/reference/html/
干货｜ Spring Cloud Stream 体系及原理介绍 https://mp.weixin.qq.com/s/e_pDTFmFcSqHH-uSIzNmMg
干货｜Spring Cloud Bus 消息总线介绍 https://mp.weixin.qq.com/s/QwIaK6LMlKyTzMKrbAEDcw
https://www.cnblogs.com/bluersw/p/11643166.html

使用场景

清空缓存：通知所有服务监听者清空某项业务的本地缓存信息，我们也可以在自定义的消息体中加业务属性，事件监听逻辑可以根据这些属性来定点清除某个特定业务对象的缓存
网关路由配置生效：在微服体系中，网关集群承载着内部服务的发现、路由、负载，当内部路由管理微服务更新了路由信息时，需要通知网关集群刷新服务路由信息

示例工程结构

├── spring-cloud-bus-demo
│   ├── pom.xml
│   ├── `spring-cloud-bus-demo-api`   //公共消息实体和事件定义
│   │   ├── pom.xml
│   │   └── src
│   ├── `spring-cloud-bus-demo-publish`  //消息发布服务
│   │   ├── pom.xml
│   │   └── src
│   ├── `spring-cloud-bus-demo-subscribe` //消息订阅服务
│   │   ├── pom.xml
│   │   └── src

file

1. 依赖

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-bus</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>

2. 定义事件

定义路由配置信息pojo

@Getter
@Setter
@Accessors(chain = true)
public class RouteInfo implements Serializable {

private static final long serialVersionUID = -6789036732613612349L;

/**
 * 服务id
 */
private String serviceId;

/**
 * 状态（1：上架；2：下架；）
 */
private Integer status;

/**
 * 路由path
 */
private String path;

/**
 * 路由uri
 */
private String uri;
}

注：要实现序列化

定义事件对象

/**
* 定义路由配置事件
*/
@Getter
@Setter
public class UserRemoteApplicationEvent extends RemoteApplicationEvent {

public UserRemoteApplicationEvent() {
}

public UserRemoteApplicationEvent(RouteInfo user, String originService, Destination destination) {
    super(user, originService, destination);
}
}

Spring Cloud bus中注册自定义事件

@SpringBootApplication(scanBasePackages = {"com.hushow.bus.demo.**"})
@RemoteApplicationEventScan(basePackageClasses = RouteRemoteApplicationEvent.class)
public class PublishApplication {
public static void main(String[] args) {
    SpringApplication.run(PublishApplication.class, args);
}
}

3. 监听事件

    @EventListener
    public void onUserRemoteApplicationEvent(RouteRemoteApplicationEvent event) {

        log.info("UserRemoteApplicationEvent:{}, Source:{}, destinationService: {}",event.getSource(),
                event.getOriginService(),
                event.getDestinationService());

        log.info("user: {}", JSONObject.toJSONString(event.getSource()));
    }

4. 发送事件

    @GetMapping("/sendMessage")
    @ApiOperation(value = "发送", notes = "发送")
    public String list(@RequestParam("message") String message) {
        RouteInfo routeInfo = new RouteInfo().setServiceId("app-one-service").setStatus(1).setUri("/one/add").setUri("lb:app-one-service");
        RouteRemoteApplicationEvent event = new RouteRemoteApplicationEvent(routeInfo, busProperties.getId(), ()->"**");
        eventPublisher.publishEvent(event);
        return message;
    }