SpringCloud 学习第十六节 消息总线 Bus

消息总线 Bus

概述

    在上一节的学习中，分布式配置项已经实现了手动刷新，那么自动刷新需要用户学习总线Bus。Spring Cloud Bus 配合Spring Cloud Config可以实现配置的动态刷新。Spring Cloud Bus是用来将分布式系统的节点与轻量级消息系统连接起来的框架。它整合了Java的事件处理机制和消息中间件的功能。Bus支持两种消息代理：RabbitMQ和Kafka。

    Spring Cloud Bus能管理和传播分布式系统间的消息，就像一个分布式执行器，可用于广播状态更改、事件推送等，也可以当作微服务间的通信通道。
    什么是总线：在微服务架构的系统中，通常会使用轻量级的消息代理来构建一个共用的消息主题，并让系统中所有微服务的实例都连接上来，由于该主题中产生的消息会被所有的实例监听和消费，所以称它为消息总线，在总线上的各个实例，都可以方便的广播一些需要让其他连接在该主题上的实例都知道的消息。ConfigClient实例都监听MQ中同一个topic,默认是(springCloudBus)。当一个服务刷新数据的时候，它会把这个信息放入到同topic中，这样其他监听同一个topic的服务就能得到通知，然后去更新自身配置

RibbonMQ 安装

    docker search rabbitmq
    docker pull rabbitmq 拉取官方的rabbitmq
    启动rabbitmq docker run -d --hostname my-rabbit --name rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123456 -p 15672:15672 -p 5672:5672 rabbitmq 用户名admin 密码123456。启动成功后访问 http://localhost:15672/#/ 如果失败，无法访问进入镜像docker exec -it 0bcee00ccbfe /bin/bash 开启 rabbitmq-plugins enable rabbitmq_management 即可访问

以3355为模版制作3366微服务

    建cloud-config-client-3366模块
    pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>zjt-cloud-api</artifactId>
        <groupId>com.zjt.cloud-api</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>cloud-config-client-3366</artifactId>

    <dependencies>
        <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>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

    bootstrap.yml

server:
  port: 3366

eureka:
  client:
    fetch-registry: true
    register-with-eureka: true
    service-url:
      defaultZone: http://eureka7001.com:7001/eureka

spring:
  application:
    name: cloud-config-client
  cloud:
    config: # config 客户端配置
      label: master #分支名称
      name: config # 配置文件名称
      profile: dev # 读取名称后缀 结合上面的我们需要读取的是 master分支上的config-dev.yml的配置文件
      uri: http://localhost:3344 # 配置中心地址 读取http://localhost:3344/master/config-dev.yml

# 暴露监控端点
management:
  endpoints:
    web:
      exposure:
        include: "*"

    主启动类

package com.zjt.cloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;

import java.util.TimeZone;

/**
 * @author zjt
 * @date 2020-09-16
 */
@EnableEurekaClient
@SpringBootApplication
public class ConfigClient3366Application {

    public static void main(String[] args) {
        // 时区设置
        TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
        SpringApplication.run(ConfigClient3366Application.class, args);
    }
    
}

    业务类

package com.zjt.cloud.config;

import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

/**
 * @author zjt
 * @date 2020-09-16
 */
@Data
@Component
@RefreshScope
public class Config {

    @Value("${config.info}")
    private String configInfo;

    @Value("${config.name}")
    private String name;

}

package com.zjt.cloud.controller;

import com.zjt.cloud.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.LinkedHashMap;
import java.util.Map;

/**
 * @author zjt
 * @date 2020-09-16
 */
@RestController
@RequestMapping("/config-client")
public class ConfigClientController {

    @Autowired
    private Config config;

    @GetMapping
    private Map<String, String> getConfigInfo() {
        return new LinkedHashMap<String, String>() {{
            put("info", config.getConfigInfo());
            put("name", config.getName());
        }};
    }

}

Bus动态刷新全局广播配置

    设计思想有两种，1、利用消息总线触发一个客户端/bus/refresh,而刷新所有客户端的配置。

    2、利用消息总线触发一个服务端的configServer的/bus/refresh端点，而刷新所有客户端的配置。图二的架构显然更加合适

    设计思想二比一更加适合，思路一不合适的原因如下，1、打破了微服务的职责单一性，因为微服务本身是业务模块，它本不应该承担配置刷新的职责。2、破坏了微服务各个节点的对等性。3、有一定的局限性，例如微服务在迁移时，它的网络地址可能会发生变化，此时如果想要做到自动刷新，那就会增加更多的修改，所以我们选择设计思想二。
    3344 pom.xml 修改添加

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

     application.yml 修改为

server:
  port: 3344

spring:
  application:
    name: cloud-config-center #注册进Eureka服务器的微服务名
  cloud:
    config:
      server:
        git:
          uri: https://github.com/JinTaoZh/zjt-cloud.git #GitHub上面的git仓库名字
          ####搜索目录
          search-paths:
            - zjt-cloud
      ####读取分支
      label: master
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: 123456


#服务注册到eureka地址
eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka

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

     3355 和 3366 修改 pom.xml 添加

<!--添加消息总线RabbitMQ支持-->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bus-amqp</artifactId>
        </dependency>

    bootstrap.yml 添加 rabbitmq

spring:
  application:
    name: cloud-config-client
  cloud:
    config: # config 客户端配置
      label: master #分支名称
      name: config # 配置文件名称
      profile: dev # 读取名称后缀 结合上面的我们需要读取的是 master分支上的config-dev.yml的配置文件
      uri: http://localhost:3344 # 配置中心地址 读取http://localhost:3344/master/config-dev.yml
  #rabbitmq相关配置 15672是Web管理界面的端口；5672是MQ访问的端口
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: 123456

    启动测试 修改 github上的配置文件




    使用命令通知3344 更新 curl -X POST "http://localhost:3344/actuator/bus-refresh"

Bus动态刷新定点通知

    使用命令 curl -X POST "http://localhost:3344/actuator/bus-refresh/cloud-config-client:3355" 定点通知3355

     3366 没有修改

     curl -X POST "http://localhost:3344/actuator/bus-refresh/cloud-config-client" 再次测试不添加端口号，两个微服务都接收到了通知。
    小总结