Spring Cloud构建微服务架构：消息驱动的微服务（消费分区）【Dalston版】

妖狐艹你老母 2021-12-05 17:05 290阅读 0赞

通过上一篇[《消息驱动的微服务（消费组）》][Link 1]的学习，我们已经能够在多实例环境下，保证同一消息只被一个消费者实例进行接收和处理。但是，对于一些特殊场景，除了要保证单一实例消费之外，还希望那些具备相同特征的消息都能够被同一个实例进行消费。这时候我们就需要对消息进行分区处理。

## 使用消息分区 ##

在Spring Cloud Stream中实现消息分区非常简单，我们可以根据消费组示例做一些配置修改就能实现，具体如下：

*  在消费者应用SinkReceiver中，我们对配置文件做一些修改，具体如下：

从上面的配置中，我们可以看到增加了这三个参数：  
1.`spring.cloud.stream.bindings.input.consumer.partitioned`：通过该参数开启消费者分区功能；

2.`spring.cloud.stream.instanceCount`：该参数指定了当前消费者的总实例数量；

3.`spring.cloud.stream.instanceIndex`：该参数设置当前实例的索引号，从0开始，最大值为 - 1。我们试验的时候需要启动多个实例，可以通过运行参数来为不同实例设置不同的索引值。（生产者那么也是可以通过  
`spring.cloud.stream.bindings.output.producer.partition-key-expression`来指定把消息发送给哪个消费者实例对象）

*  在生产者应用`SinkSender`中，我们对配置文件也做一些修改，具体如下：

spring.cloud.stream.bindings.output.destination=greetings
    spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload
    spring.cloud.stream.bindings.output.producer.partitionCount=2

从上面的配置中，我们可以看到增加了这两个参数：

1.`spring.cloud.stream.bindings.output.producer.partitionKeyExpression`：通过该参数指定了分区键的表达式规则，我们可以根据实际的输出消息规则来配置SpEL来生成合适的分区键；

2.`spring.cloud.stream.bindings.output.producer.partitionCount`：该参数指定了消息分区的数量。

到这里消息分区配置就完成了，我们可以再次启动这两个应用，同时消费者启动多个，但需要注意的是要为消费者指定不同的实例索引号，这样当同一个消息被发给消费组时，我们可以发现只有一个消费实例在接收和处理这些相同的消息。

本文转载地址：[http://blog.didispace.com/spring-cloud-starter-dalston-7-4/][http_blog.didispace.com_spring-cloud-starter-dalston-7-4]

**如果看不懂，那么下面这个例子一定可以看得懂**

### Spring Cloud Stream + RabbitMQ 消息消费组与消息分区 ###

在本 DEMO中有三个节点，stream-hello2 是消息生产端，stream-hello 和 stream-hello3 是消息消费端。

一、stream-hello2 节点（消息生产端）

1.  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">
    	<modelVersion>4.0.0</modelVersion>
     
    	<groupId>com.sande</groupId>
    	<artifactId>stream-hello2</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<packaging>jar</packaging>
     
    	<name>stream-hello2</name>
    	<description>Demo project for Spring Boot</description>
     
    	<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>2.0.4.RELEASE</version>
    		<relativePath/> 
    	</parent>
     
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    		<java.version>1.8</java.version>
    		<spring-cloud.version>Finchley.SR1</spring-cloud.version>
    	</properties>
     
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web-services</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.cloud</groupId>
    			<artifactId>spring-cloud-config-server</artifactId>
    		</dependency>
    		
     
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    		  
    		<dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
                <version>2.0.1.RELEASE</version>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
     
         
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-starter-bus-amqp</artifactId>
            </dependency>
            
            
            <dependency>
               <groupId>com.fasterxml.jackson.core</groupId>
               <artifactId>jackson-annotations</artifactId>
               <version>2.9.6</version>
            </dependency>
        
            <dependency>
               <groupId>com.fasterxml.jackson.core</groupId>
               <artifactId>jackson-core</artifactId>
               <version>2.9.6</version>
            </dependency>
        
            <dependency>
               <groupId>com.fasterxml.jackson.core</groupId>
               <artifactId>jackson-databind</artifactId>
               <version>2.9.6</version>
            </dependency>
        
    	</dependencies>
     
    	<dependencyManagement>
    		<dependencies>
    			<dependency>
    				<groupId>org.springframework.cloud</groupId>
    				<artifactId>spring-cloud-dependencies</artifactId>
    				<version>${spring-cloud.version}</version>
    				<type>pom</type>
    				<scope>import</scope>
    			</dependency>
    		</dependencies>
    	</dependencyManagement>
     
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.springframework.boot</groupId>
    				<artifactId>spring-boot-maven-plugin</artifactId>
    			</plugin>
    		</plugins>
    	</build>
     
     
    </project>

1.  src/main/resources/application.properties

spring.cloud.stream.bindings.output.destination=greetings
    server.port=8081
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

配置了spring.cloud.stream.bindings.output.destination=greetings 后会在RabbitMQ 中创建一个名为 greetings 交换器（exchange）。spring.cloud.stream.bindings.output.destination=greetings 的意思是把 spring cloud stream 的消息输出通道绑定到 RabbitMQ 的 greetings 交换器。

1.  消息生产类

package com.sande.streamhello.ramp;
     
    import java.text.SimpleDateFormat;
    import java.util.Date;
     
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    import org.springframework.cloud.stream.messaging.Source;
    import org.springframework.context.annotation.Bean;
    import org.springframework.integration.annotation.InboundChannelAdapter;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.support.GenericMessage;
    import org.springframework.integration.annotation.Poller;
    import org.springframework.integration.core.MessageSource;
     
    @EnableBinding(value= {Source.class})
    public class SinkSender {
    	     private static Logger logger = LoggerFactory.getLogger(SinkSender.class);
            
    	     //private String format="yyyy-mm-dd  HH:mm:ss";
    	     private Integer i=0;
    	     @Bean
    	     @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000", maxMessagesPerPoll = "1"))
    	     public MessageSource<Integer> timerMessageSource() {
    	    	    System.out.println("MessageSource");
    	    	    //return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
    	    	    //return () -> new GenericMessage<>("wo ai ni wo de jia");
    	    	    //return () -> new GenericMessage<>("{\"name\":\"didi\",\"age\":30}");
    	    	    
    	    	    return () -> new GenericMessage<>(i++);
    	    	  }
    }

1.  应用主类

package com.sande.streamhello;
     
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
     
     
    @SpringBootApplication
    public class StreamHelloApplication {
     
    	public static void main(String[] args) {
    		SpringApplication.run(StreamHelloApplication.class, args);
    	}
    }

二、stream-hello 节点（消息消费端）

1.  pom.xml

1.  src/main/resources/application.properties

spring.cloud.stream.bindings.input.group=Service-A
    spring.cloud.stream.bindings.input.destination=greetings
    server.port=8080
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

配置了 spring.cloud.stream.bindings.input.destination=greetings 后会在RabbitMQ 中创建一个名为 greetings 交换器（exchange）。spring.cloud.stream.bindings.input.destination=greetings 的意思是把 spring cloud stream 的输入通道绑定到 RabbitMQ 的 greetings 交换器。这样节点 stream-hello 的输入通道对应节点 stream-hello2 的输出通道，stream-hello 节点就配置成了 stream-hello2 节点的消费端。spring.cloud.stream.bindings.input.group=Service-A 配置 stream-hello 为消息组 Service-A 中的一个消费端。这两个配置项联合起来解释，就是把节点 stream-hello 的输入通道绑定到 RabbitMQ 的 greetings 交换器，并设置为greetings 交换器中 Service-A 消息消费组中的消费端节点。

1.  消息消费类

package com.sande.streamhello.ramp;
     
    import java.util.stream.Stream;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    import com.sande.streamhello.StreamHelloApplication;
     
    @EnableBinding(value= {Sink.class})
    public class SinkReceiver {
    	
    	private static Logger logger = LoggerFactory.getLogger(StreamHelloApplication.class);
    	
    	@StreamListener(Sink.INPUT)
    	public void receive(String payload) {
    		logger.info("Received:" + payload);
    	}
    }

四、测试

1.  启动RabbitMQ
2.  启动节点 stream-hello 和 stream-hello3
3.  启动节点 stream-hello2
4.  我们看到 RabbitMQ 中已经创建了 greetings 交换器  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70]
5.  我们看到 RabbitMQ 中已经创建了greetings.Service-A 消息队列  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 1]
6.  消费节点接收到的消息

6.1 stream-hello 节点接收的消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 2]  
6.2 stream-hello3 接收到的消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 3]  
通过两个消费端输出的消息我们看到，每条消息只会被其中一个消费节点接收。

五、配置消息

1.  消息生产端 src/main/resources/application.properties

spring.cloud.stream.bindings.output.destination=greetings
    spring.cloud.stream.bindings.output.producer.partition-count=2
    spring.cloud.stream.bindings.output.producer.partition-key-expression=1
    server.port=8081
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

spring.cloud.stream.bindings.output.producer.partition-count=2 指定参与消息分区的消费端节点数量为2个。

spring.cloud.stream.bindings.output.producer.partition-key-expression=1 表示只有分区ID为1的消费端能接收到信息。

spring.cloud.stream.bindings.output.producer.partition-key-expression=0 表示只有分区ID为0的消费端能接收到信息。

2.消费端 stream-hello src/main/resources/application.properties

spring.cloud.stream.bindings.input.group=Service-A
    spring.cloud.stream.bindings.input.destination=greetings
    spring.cloud.stream.bindings.input.consumer.partitioned=true
    spring.cloud.stream.instance-count=2
    spring.cloud.stream.instance-index=0
    server.port=8080
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

spring.cloud.stream.bindings.input.consumer.partitioned=true 表示启用消息分区功能

spring.cloud.stream.instance-count=2 表示消息分区的消费端节点数量为2个

spring.cloud.stream.instance-index=0 该参数设置消费端实例的索引号，索引号从0开始。这里设置该节点的索引号为0

3.消费端 stream-hello3 src/main/resources/application.properties

spring.cloud.stream.bindings.input.consumer.partitioned=true
    spring.cloud.stream.instance-count=2
    spring.cloud.stream.instance-index=1

1.  测试

4.1 重启 stream-hello 、sream-hello2、stream-hello3

4.2 当消息生产端 设置 spring.cloud.stream.bindings.output.producer.partition-key-expression=1 时，只有实例索引号为1的 stream-hello3 节点可以接收到信息。从截图我们看到 stream-hello3 接收到的消息的数字都是连续的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 4]  
4.3 当消息生产端 设置 spring.cloud.stream.bindings.output.producer.partition-key-expression=0 时，只有实例索引号为0的 stream-hello 节点可以接收到信息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 5]  
查看 RabbitMQ Management 可以看到，其实在 RabbitMQ 中，针对两个消费端分别创建了消息队列。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 6]  
本文转载地址：[https://blog.csdn.net/lixiaxin200319/article/details/82930026][https_blog.csdn.net_lixiaxin200319_article_details_82930026]

[Link 1]: https://blog.csdn.net/qq_40241957/article/details/96282944
[http_blog.didispace.com_spring-cloud-starter-dalston-7-4]: http://blog.didispace.com/spring-cloud-starter-dalston-7-4/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70]: /images/20211205/7c7eef46c45446bd91125bc179539c64.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 1]: /images/20211205/d87a72d3f77347d89f8f76d9fab7c9c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 2]: /images/20211205/3cabe91d3a7c4702a8a76a3d28afbb36.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 3]: /images/20211205/f319e21f7f3b4fa8945cfa7fde1c3426.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 4]: /images/20211205/4653baca68ed4325bb2cefe2750f493c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 5]: /images/20211205/fa48cf0a02dc4b33a9c75e34bdf15beb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjQxOTU3_size_16_color_FFFFFF_t_70 6]: /images/20211205/791775b238b34e2ea60e2b4d2b744ccc.png
[https_blog.csdn.net_lixiaxin200319_article_details_82930026]: https://blog.csdn.net/lixiaxin200319/article/details/82930026