消息代理

消息代理是一种消息验证、传输、路由的架构模式。

RabbitMQ实现消息总线

消息投递到队列的整个过程如下：
1.客户端连接到消息队列服务器，打开一个Channel
2.客户端声明一个Exchange，并设置相关属性
3.客户端声明一个Queue,并设置相关属性
4.客户端使用RoutingKey，在Exchange和Queue之间建立好绑定关系
5.客户端投递消息到Exchange
6.Exchange接收到消息后，根据消息的Key和已经设置的Binding，进行消息路由，将消息投递到一个或多个Queue中
Exchange有几种类型：
1.Direct交换机:完全根据Key进行投递
2.Topic交换机:对Key进行模式匹配后进行投递，可以使用符号#匹配一个或多个词，符号*匹配正好一个词
3.Fanout交换机:不需要任何Key,它采取广播的模式，一个消息进来时，投递到与该交换机绑定的所有队列

快速入门

在Spring Boot中整合RabbitMQ,pom中AMQP模块可以很好地支持RabbitMQ

• 新建一个Spring Boot工程，命名为rabbitmq-hello
• 在pom.xml中引入如下依赖内容，其中spring-boot-starter-amqp用于支持RabbitMQ

<dependencies>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-amqp</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-test</artifactId>
		<scope>test</scope>
	</dependency>
</dependencies>

• 此处需要在RabbitMQ的管理页面http://localhost:15672/#/users添加用户,比如springcloud
  
• 在application.properties中配置关于RabbitMQ的连接和用户信息

spring.application.name=rabbitmq-hello
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=springcloud
spring.rabbitmq.password=123456

• 创建消息生产者Sender

@Component
public class Sender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void send(){
        String context = "hello" + new Date();
        System.out.println("Sender: " + context);
        this.rabbitTemplate.convertAndSend("hello",context);
    }
}

• 创建消息消费者Receiver

@Component
@RabbitListener(queues = "hello")
public class Receiver {
    @RabbitHandler
    public void process(String hello){
        System.out.println("Receiver: " + hello);
    }
}

• 创建RabbitMQ的配置类RabbitConfig，用来配置队列，交换器、路由等高级信息

import org.springframework.amqp.core.Queue;
@Configuration
public class RabbitConfig {
    @Bean
    public Queue helloQueue(){
        return new Queue("hello");
    }
}

• 创建应用主类

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

• 创建单元测试类，用来调用消息生产

@RunWith(SpringRunner.class)
@SpringBootTest
public class HelloApplicationTests {
	@Autowired
	private Sender sender;
	@Test
	public void hello() throws Exception{
		sender.send();
	}
}

• 启动主类，从控制台中，可以看到

org.springframework.amqp.AmqpConnectException: java.net.ConnectException: Connection refused (Connection refused)

这里不得不提下这个问题，书中没有提到这个问题，所以自己想办法解决了，首先打开RabbitMQ的管理界面可以看到springcloud处于No access状态

给springcloud用户添加权限，单击springcloud，进入下图
单击Set permission，此时springcloud用户获取到权限了

• 重新启动应用，可以看到控制台

Created new connection: rabbitConnectionFactory#2b608c2a:0/SimpleConnection@5e53c92d [delegate=amqp://springcloud@127.0.0.1:5672/, localPort= 49700]

• 通过RabbitMQ的管理页面可以看到连接的条目
  
  
• 运行单元测试类，消息被发送到了RabbitServer的队列中
• 切换到应用主类的控制台，可以看到消费者对队列的监听程序执行了，并输出了接收到的消息信息
  

整合Spring Cloud Bus

• 扩展config-client-eureka的应用
  修改pom.xml，增加spring-cloud-starter-bus-amqp模块

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

• 在配置文件中增加关于RabbitMQ的连接和用户信息

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=springcloud
spring.rabbitmq.password=123456

• 启动config-server,再启动两个config-client，这里启动两个的目的就是达到多个客户端同时刷新的效果
• 先访问两个config-client的/from请求，会返回当前dongarea-dev.properties中的from属性，比如为git-dev-1.0
• 再修改dongarea-dev.properties的值,比如为git-dev-2.0,然后用POSTMAN来发送POST请求到其中一个/bus/refresh,返回的是401的异常信息
  
• 查看pom中的spring-cloud-starter-bus-amqp的版本为1.3.4.RELEASE版，经百度查询，需要去除安全验证，在bootstrap.properties中添加management.security.enabled=false,再次执行更新，显示成功
  
• 再次执行两个config-client的/from请求，返回git-dev-2.0，显示更新成功
  这样就完成了Spring Cloud Bus来实时更新总线上的属性配置了

原理分析

这里主要是通过消息总线的方式，将其中一个实例的更新请求/bus/refresh，传递给其他实例，达到同时刷新的目的

指定刷新范围

特别情况下，希望可以刷新微服务中某个具体实例的配置
通过/bus/refresh?destination=customers:9000来刷新某个具体的实例，customers:9000指的是各个微服务的ApplicationContext ID
默认情况下，ApplicationContext ID是spring.application.name:server.port，详见org.springframework.boot.context.ContextIdApplicationContextInitializer.getApplicationId(ConfigurableEnvironment)方法。
destination参数也可以用来定位特定的微服务。例如：/bus/refresh?destination=customers:**，这样就可以触发customers微服务所有实例的配置刷新。

架构优化

在之前的架构中，服务是不对等的，如果固定向某个实例发送刷新请求，再触发整个服务集群的刷新，这种方式会造成运维工作的复杂性。若此服务实例发生迁移，则会造成配置不能刷新的潜在后果。
架构调整可以做如下改动：
1.在Config Server中也引入Spring Cloud Bus,将配置服务端也加入到消息总线中来
2./bus/refresh请求不再发送到具体服务实例上，而是发送给Config Server，并通过
destination参数来指定需要更新配置的服务实例
对于Git的触发等配置只需要针对Config Server即可，从而简化了集群上的一些维护工作