SpringCloud——消息总线 SpringCloud Bus

程序员文章站 2022-06-12 19:29:29

...

1. 概述

Spring Cloud Bus 配合 Spring Cloud Config 使用可以实现配置的动态刷新。

Spring Cloud Bus 是用来将分布式系统的节点与轻量级消息系统链接起来的框架，它整合了 Java 的事件处理机制和消息中间件的功能。

Spring Clud Bus目前支持 RabbitMQ 和 Kafka。

SpringCloud——消息总线 SpringCloud Bus

Spring Cloud Bus 能管理和传播分布式系统间的消息,就像一个分布式执行器，可用于广播状态更改、事件推送等，也可以当作微服务间的通信通道

SpringCloud——消息总线 SpringCloud Bus

什么是总线：

在微服务架构的系统中，通常会使用轻量级的消息代理来构建一个共用的消息主题，并让系统中所有微服务实例都连接上来。由于该主题中产生的消息会被所有实例监听和消费，所以称它为消息总线。在总线上的各个实例，都可以方便地广播一些需要让其他连接在该主题上的实例都知道的消息。

基本原理：

ConfigClient 实例都监听MQ中同一个topic(默认是springCloudBus)。当一个服务刷新数据的时候，它会把这个信息放入到Topic中,这样其它监听同一Topic的服务就能得到通知，然后去更新自身的配置。

2. 动态刷新全局广播

2.1 设计思路

思路一：利用消息总线触发一个客户端/bus/refresh，而刷新所有客户端的配置（图1 模式）

思路二：利用消息总线触发一个服务端 ConfigServer 的 /bus/refresh 端点，而刷新所有客户端的配置（图2模式）

思路一打破了微服务的职责单一性，因为微服务本身是业务模块，它本不应该承担配置刷新职责，破坏了微服务各节点的对等性。所以选用思路二。

2.2 实现

准备三个微服务（一个服务端3344，两个客户端3355、3366）。

SpringCloud——消息总线 SpringCloud Bus

2.2.1 服务端

pom:

<dependencies>
	<!--添加消息总线支持-->
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-bus-amqp</artifactId>
	</dependency>
	<!--分布式配置支持-->
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-config-server</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-web</artifactId>
	</dependency>
	<!--监控支持-->
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-actuator</artifactId>
	</dependency>
</dependencies>

application.yml:

server:
  port: 3344
spring:
  application:
    name: cloud-config-center
  cloud:
    config:
      server:
        git:
          uri:  aaa@qq.com:Zhangtao153/springcloud-config.git
          search-paths:
            - springcloud-config
      label: master
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

eureka:
  client:
    service-url:
      defaultZone:  http://localhost:7001/eureka

# 暴露 bus 刷新配置的端点
management:
  endpoints:
    web:
      exposure:
        include: 'bus-refresh'

启动类：

@SpringBootApplication
@EnableConfigServer
public class ConfigCenterMain3344 {
    public static void main(String[] args){
        SpringApplication.run(ConfigCenterMain3344.class,args);
    }
}

2.2.2 客户端

pom:

<dependencies>
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-bus-amqp</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-config</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-web</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-actuator</artifactId>
	</dependency>
</dependencies>

bootstrap.yml:

server:
  port: 3355

spring:
  application:
    name: config-client
  cloud:
    config:
      label: master  # 分支
      name: config   # 配置文件名
      profile: dev   # 环境
      uri: http://localhost:3344
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka

management:
  endpoints:
    web:
      exposure:
        include: "*"

启动类：

@SpringBootApplication
public class ConfigClientMain3355 {
    public static void main(String[] args){
        SpringApplication.run(ConfigClientMain3355.class,args);
    }
}

业务类：

@RestController
@RefreshScope
public class ConfigClientController {
    @Value("${server.port}")
    private String serverPort;

    @Value("${config.info}")
    private String configInfo;
    
    @GetMapping("/configInfo")
    public String getConfigInfo(){
        return "serverPort:"+serverPort+"\t\n\n configInfo: "+configInfo;
    }
}