Spring Cloud学习day106：消息总线Bus和消息驱动Stream

一、Spring Cloud Bus介绍

1.什么是Spring Cloud Bus？

Spring Cloud Bus集成了市面上常用的消息代理等，连接微服务系统中所有的节点，当有数据变更时，可以通过消息广播通知服务及时变更数据。

解决了什么问题：

解决了微服务数据变更，及时同步的问题。

Clien刷新当Service-A获取到配置信息后会向RabbitMq发送消息，mq在通知其他两个客户端，获取最新的配置信息

Server刷新配置中心会向RabbitMQ发送刷新的通知，RabbitMQ会向所有的客户端发送更新的消息，客户端再从服务中获取最新的配置信息

2.使用Bus实现自动刷新配置信息-Client刷新：

需要安装RabbitMq；https://www.jianshu.com/p/da5b3b224db2

创建配置中心Server服务：

示例
修改POM文件：

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-config-server</artifactId>
        </dependency>
    </dependencies>

修改配置文件：

spring.application.name=config-server
server.port=9020

#设置服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://admin:123456@eureka1:8761/eureka/,http://admin:123456@eureka2:8761/eureka/

#Git 配置 
spring.cloud.config.server.git.uri=https://gitee.com/zwzy/config
#Git私有仓库的用户名和密码
#spring.cloud.config.server.git.username=
#spring.cloud.config.server.git.password=

修改启动类：

@EnableConfigServer
@EnableEurekaClient
@SpringBootApplication
public class ConfigServerApplication {

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

创建客户端项目：

示例
修改POM文件：

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-config</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bus-amqp</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

修改配置文件：

spring.application.name=config-client
server.port=9022

#设置服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://admin:123456@eureka1:8761/eureka/,http://admin:123456@eureka2:8761/eureka/

#默认 false,这里设置 true,表示开启读取配置中心的配置
spring.cloud.config.discovery.enabled=true
#对应 eureka 中的配置中心 serviceId，默认是configserver
spring.cloud.config.discovery.serviceId=config-server
#指定环境
spring.cloud.config.profile=dev
#git标签 分支/主干
spring.cloud.config.label=master

#springboot 默认开启了权限拦截 会导致 /refresh 出现 401，拒绝访问 
management.security.enabled=false

#消息队列的链接配置 
spring.rabbitmq.host=192.168.226.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtualHost=/

修改启动类：

@EnableEurekaClient
@SpringBootApplication
public class RefreshApplication {

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

创建Controller：

@RestController
@RefreshScope//刷新作用域
public class ConfigClientController {
    
    @Value("${e-book}")
    private String msg;
    
    @RequestMapping("/show")
    public String ShowMsg() {
        return this.msg;
    }
}

测试：

远程仓库

示例

拷贝Bus客户端项目：

示例
修改配置文件，更改端口号：

spring.application.name=config-client
server.port=9021

#设置服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://admin:123456@eureka1:8761/eureka/,http://admin:123456@eureka2:8761/eureka/

#默认 false,这里设置 true,表示开启读取配置中心的配置
spring.cloud.config.discovery.enabled=true
#对应 eureka 中的配置中心 serviceId，默认是configserver
spring.cloud.config.discovery.serviceId=config-server
#指定环境
spring.cloud.config.profile=dev
#git标签 分支/主干
spring.cloud.config.label=master

#springboot 默认开启了权限拦截 会导致 /refresh 出现 401，拒绝访问 
management.security.enabled=false

#消息队列的链接配置 
spring.rabbitmq.host=192.168.226.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtualHost=/

使用HttpClientUtil工具发送post请求：

刷新任意客户端发送post请求

任意一个客户端发送post请求

服务端会更新本地仓库
测试：

修改远程仓库中的配置信息

示例

3.使用Bus实现自动刷新配置信息-Server 刷新：

创建服务端项目：

示例
修改POM文件：

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</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-bus-amqp</artifactId>
        </dependency>
    </dependencies>

修改配置文件：

spring.application.name=config-server
server.port=9020

#设置服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://admin:123456@eureka1:8761/eureka/,http://admin:123456@eureka2:8761/eureka/

#Git 配置 
spring.cloud.config.server.git.uri=https://gitee.com/zwzy/config
#Git私有仓库的用户名和密码
#spring.cloud.config.server.git.username=
#spring.cloud.config.server.git.password=

#springboot 默认开启了权限拦截 会导致 /refresh 出现 401，拒绝访问 
management.security.enabled=false

spring.rabbitmq.host=192.168.226.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtualHost=/

修改启动类：

@EnableConfigServer
@EnableEurekaClient
@SpringBootApplication
public class ConfigServerApplication {

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

使用httpClientUtil工具发送post请求：

示例

使用配置中心的服务端发送post请求

测试：

客户端使用上面的两个项目。

修改git中配置的信息

9021

9022

4.实现局部刷新服务：

使用上面的Server-bus2服务和两个客户端测试。

1.刷新指定服务：

使用httpClientUtil工具类发送post请求。
“http://Config-Server/bus/refresh?destination=需要刷新的服务名称:端口 ”。

修改远程仓库中的配置信息：

示例
使用httpClientUtil工具类：

示例
测试：

9021

9022

2.刷新指定集群：

使用工具类发送post请求；
"http://Config-Server/bus/refresh?destination=需要刷新的服务名称:**"。

修改远程仓库中配置信息：

示例
使用工具类发送post请求：

示例
测试：

9022

9021

二、消息驱动入门案例

1.创建消息的发送者：

示例

修改POM文件：

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</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-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

修改配置文件：

spring.application.name=config-server
server.port=9020

#设置服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://admin:123456@eureka1:8761/eureka/,http://admin:123456@eureka2:8761/eureka/

#消息队列的链接配置 
spring.rabbitmq.host=192.168.226.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtualHost=/

创建发送消息的接口：

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.SubscribableChannel;

public interface ISendeService {

    @Output("user-exchange")
    SubscribableChannel send(); 
}

修改启动类：

@EnableBinding(value = ISendeService.class)
@EnableEurekaClient
@SpringBootApplication
public class ISendeApplication {

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

2.创建消息的接受者：

示例

修改POM文件：

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Dalston.SR5</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</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-stream-rabbit</artifactId>
        </dependency>
    </dependencies>

修改配置文件：

spring.application.name=config-server
server.port=9021

#设置服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://admin:123456@eureka1:8761/eureka/,http://admin:123456@eureka2:8761/eureka/

#消息队列的链接配置 
spring.rabbitmq.host=192.168.226.129
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtualHost=/

创建接收消息的接口：

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface IReceiveService {

    @Input("user-exchange")
    SubscribableChannel receiver(); 
}

创建处理消息的类：

@Service
@EnableBinding({IReceiveService.class})
public class ReceiverService {

    @StreamListener("user-exchange")
    public void onReciver(byte[] msg) {
        //处理消息
        System.out.println("Receiver："+new String(msg));
    }
}

修改启动类：

@EnableBinding(value = IReceiveService.class)
@EnableEurekaClient
@SpringBootApplication
public class IReceiverApplication {

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

测试：

在消息发送的项目中创建测试类；依次启动项目和执行测试类。

@RunWith(SpringRunner.class)
@SpringBootTest(classes = ISendeApplication.class)
public class Test {
    
    @Autowired
    private ISendeService isendeService;
    
    @org.junit.Test
    public void testSend() {
        String msg ="Sende发送的消息！";
        //将消息封装成Message
        Message message = MessageBuilder.withPayload(msg.getBytes()).build();
        isendeService.send().send(message);
    }
}

示例