Spring cloud stream【消息分区】

- 日理万妓 2021-12-20 06:23 192阅读 0赞

在上篇文章中我们给大家介绍了Stream的消息分组，可以实现消息的重复消费的问题，但在某些场景下分组还不能满足我们的需求，比如，同时有多条同一个用户的数据，发送过来，我们需要根据用户统计，但是消息被分散到了不同的集群节点上了，这时我们就可以考虑**消息分区**了。  
  当生产者将消息数据发送给多个消费者实例时，保证同一消息数据始终是由同一个消费者实例接收和处理。

# Stream 消息分区 #

## 创建项目 ##

将我们上篇文章中的分组的三个项目，拷贝一份修改名称及服务名称

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]

## 没有分区的情况下演示 ##

发送多条消息查看效果

@RunWith(SpringRunner.class)
    @SpringBootTest(classes=StreamSenderStart.class)
    public class StreamTest { 
    	
    	@Autowired
    	private ISendeService sendService;
    
    	@Test
    	public void testStream(){ 
    		Product p = new Product(999, "stream test ...999");
    		// 将需要发送的消息封装为Message对象
    		Message message = MessageBuilder
    								.withPayload(p)
    								.build();
    		for (int i = 0; i < 10; i++) { 
    			// 发送多条消息到队列中
    			sendService.send().send(message );
    		}
    		
    	}
    }

10条消息被随机的分散到了两个消费者中:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 2]  
我们可以看到A中6条消息，B中4条消息，而且这是随机的，下次执行的结果肯定不一样。

## 分区 ##

### 1.发送者中配置 ###

spring.application.name=stream-partition-sender
    server.port=9060
    #设置服务注册中心地址，指向另一个注册中心
    eureka.client.serviceUrl.defaultZone=http://dpb:123456@eureka1:8761/eureka/,http://dpb:123456@eureka2:8761/eureka/
    
    #rebbitmq 链接信息
    spring.rabbitmq.host=192.168.88.150
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=dpb
    spring.rabbitmq.password=123
    spring.rabbitmq.virtualHost=/
    
    # 对应 MQ 是 exchange  outputProduct自定义的信息
    spring.cloud.stream.bindings.outputProduct.destination=exchangeProduct
    
    #通过该参数指定了分区键的表达式规则
    spring.cloud.stream.bindings.outputProduct.producer.partitionKeyExpression=payload
    #指定了消息分区的数量。 
    spring.cloud.stream.bindings.outputProduct.producer.partitionCount=2

### 2.消费者中配置 ###

服务A

spring.application.name=stream-partition-receiverA
    server.port=9070
    #设置服务注册中心地址，指向另一个注册中心
    eureka.client.serviceUrl.defaultZone=http://dpb:123456@eureka1:8761/eureka/,http://dpb:123456@eureka2:8761/eureka/
    
    #rebbitmq 链接信息
    spring.rabbitmq.host=192.168.88.150
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=dpb
    spring.rabbitmq.password=123
    spring.rabbitmq.virtualHost=/
    
    # 对应 MQ 是 exchange  和消息发送者的 交换器是同一个
    spring.cloud.stream.bindings.inputProduct.destination=exchangeProduct
    # 具体分组 对应 MQ 是 队列名称 并且持久化队列  inputProduct 自定义
    spring.cloud.stream.bindings.inputProduct.group=groupProduct999
    
    #开启消费者分区功能
    spring.cloud.stream.bindings.inputProduct.consumer.partitioned=true
    #指定了当前消费者的总实例数量
    spring.cloud.stream.instanceCount=2 
    #设置当前实例的索引号，从 0 开始
    spring.cloud.stream.instanceIndex=0

服务B

spring.application.name=stream-partition-receiverB
    server.port=9071
    #设置服务注册中心地址，指向另一个注册中心
    eureka.client.serviceUrl.defaultZone=http://dpb:123456@eureka1:8761/eureka/,http://dpb:123456@eureka2:8761/eureka/
    
    #rebbitmq 链接信息
    spring.rabbitmq.host=192.168.88.150
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=dpb
    spring.rabbitmq.password=123
    spring.rabbitmq.virtualHost=/
    
    # 对应 MQ 是 exchange  和消息发送者的 交换器是同一个
    spring.cloud.stream.bindings.inputProduct.destination=exchangeProduct
    # 具体分组 对应 MQ 是 队列名称 并且持久化队列  inputProduct 自定义
    spring.cloud.stream.bindings.inputProduct.group=groupProduct999
    
    #开启消费者分区功能
    spring.cloud.stream.bindings.inputProduct.consumer.partitioned=true
    #指定了当前消费者的总实例数量
    spring.cloud.stream.instanceCount=2 
    #设置当前实例的索引号，从 1 开始
    spring.cloud.stream.instanceIndex=1

启动服务测试

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 3]

10个消息都被消费者A给消费了，说明到达了我们需要的效果。  
案例源码:[https://github.com/q279583842q/springcloud-e-book][https_github.com_q279583842q_springcloud-e-book]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20211220/e2d188f224ba48dfb41d8674b589b00d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]: /images/20211220/b41ed28886644605a29a5dd1a88d57d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 2]: /images/20211220/398e1bb1b9534d1e874dfb923f45b874.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 3]: /images/20211220/faa3e696a3cb4d978b5671d44d47d976.png
[https_github.com_q279583842q_springcloud-e-book]: https://github.com/q279583842q/springcloud-e-book