devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：我家小花儿 | 2024-07-17 15:58:56

赞

踩

springboot rabbitmq分区

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

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

1. pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3. 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4. 	<modelVersion>4.0.0</modelVersion>
  5.  
  6. 	<groupId>com.sande</groupId>
  7. 	<artifactId>stream-hello2</artifactId>
  8. 	<version>0.0.1-SNAPSHOT</version>
  9. 	<packaging>jar</packaging>
  10.  
  11. 	<name>stream-hello2</name>
  12. 	<description>Demo project for Spring Boot</description>
  13.  
  14. 	<parent>
  15. 		<groupId>org.springframework.boot</groupId>
  16. 		<artifactId>spring-boot-starter-parent</artifactId>
  17. 		<version>2.0.4.RELEASE</version>
  18. 		<relativePath/> <!-- lookup parent from repository -->
  19. 	</parent>
  20.  
  21. 	<properties>
  22. 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  23. 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  24. 		<java.version>1.8</java.version>
  25. 		<spring-cloud.version>Finchley.SR1</spring-cloud.version>
  26. 	</properties>
  27.  
  28. 	<dependencies>
  29. 		<dependency>
  30. 			<groupId>org.springframework.boot</groupId>
  31. 			<artifactId>spring-boot-starter-web-services</artifactId>
  32. 		</dependency>
  33. 		<dependency>
  34. 			<groupId>org.springframework.cloud</groupId>
  35. 			<artifactId>spring-cloud-config-server</artifactId>
  36. 		</dependency>
  37. 		
  38.  
  39. 		<dependency>
  40. 			<groupId>org.springframework.boot</groupId>
  41. 			<artifactId>spring-boot-starter-test</artifactId>
  42. 			<scope>test</scope>
  43. 		</dependency>
  44. 		  
  45. 		<dependency>
  46.             <groupId>org.springframework.cloud</groupId>
  47.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  48.             <version>2.0.1.RELEASE</version>
  49.         </dependency>
  50.         
  51.         <dependency>
  52.             <groupId>org.springframework.boot</groupId>
  53.             <artifactId>spring-boot-starter-amqp</artifactId>
  54.         </dependency>
  55.  
  56.      
  57.         <dependency>
  58.             <groupId>org.springframework.cloud</groupId>
  59.             <artifactId>spring-cloud-starter-bus-amqp</artifactId>
  60.         </dependency>
  61.         
  62.         
  63.         <dependency>
  64.            <groupId>com.fasterxml.jackson.core</groupId>
  65.            <artifactId>jackson-annotations</artifactId>
  66.            <version>2.9.6</version>
  67.         </dependency>
  68.     
  69.         <dependency>
  70.            <groupId>com.fasterxml.jackson.core</groupId>
  71.            <artifactId>jackson-core</artifactId>
  72.            <version>2.9.6</version>
  73.         </dependency>
  74.     
  75.         <dependency>
  76.            <groupId>com.fasterxml.jackson.core</groupId>
  77.            <artifactId>jackson-databind</artifactId>
  78.            <version>2.9.6</version>
  79.         </dependency>
  80.     
  81. 	</dependencies>
  82.  
  83. 	<dependencyManagement>
  84. 		<dependencies>
  85. 			<dependency>
  86. 				<groupId>org.springframework.cloud</groupId>
  87. 				<artifactId>spring-cloud-dependencies</artifactId>
  88. 				<version>${spring-cloud.version}</version>
  89. 				<type>pom</type>
  90. 				<scope>import</scope>
  91. 			</dependency>
  92. 		</dependencies>
  93. 	</dependencyManagement>
  94.  
  95. 	<build>
  96. 		<plugins>
  97. 			<plugin>
  98. 				<groupId>org.springframework.boot</groupId>
  99. 				<artifactId>spring-boot-maven-plugin</artifactId>
  100. 			</plugin>
  101. 		</plugins>
  102. 	</build>
  103.  
  104.  
  105. </project>

2. src/main/resources/application.properties

  1. spring.cloud.stream.bindings.output.destination=greetings
  2. server.port=8081
  3. spring.rabbitmq.host=localhost
  4. spring.rabbitmq.port=5672
  5. spring.rabbitmq.username=guest
  6. 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 交换器。

3. 消息生产类

  1. package com.sande.streamhello.ramp;
  2.  
  3. import java.text.SimpleDateFormat;
  4. import java.util.Date;
  5.  
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8. import org.springframework.cloud.stream.annotation.EnableBinding;
  9. import org.springframework.cloud.stream.annotation.Output;
  10. import org.springframework.cloud.stream.annotation.StreamListener;
  11. import org.springframework.cloud.stream.messaging.Sink;
  12. import org.springframework.cloud.stream.messaging.Source;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.integration.annotation.InboundChannelAdapter;
  15. import org.springframework.messaging.MessageChannel;
  16. import org.springframework.messaging.support.GenericMessage;
  17. import org.springframework.integration.annotation.Poller;
  18. import org.springframework.integration.core.MessageSource;
  19.  
  20. @EnableBinding(value= {Source.class})
  21. public class SinkSender {
  22. 	     private static Logger logger = LoggerFactory.getLogger(SinkSender.class);
  23.         
  24. 	     //private String format="yyyy-mm-dd  HH:mm:ss";
  25. 	     private Integer i=0;
  26. 	     @Bean
  27. 	     @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "2000", maxMessagesPerPoll = "1"))
  28. 	     public MessageSource<Integer> timerMessageSource() {
  29. 	    	    System.out.println("MessageSource");
  30. 	    	    //return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
  31. 	    	    //return () -> new GenericMessage<>("wo ai ni wo de jia");
  32. 	    	    //return () -> new GenericMessage<>("{\"name\":\"didi\",\"age\":30}");
  33. 	    	    
  34. 	    	    return () -> new GenericMessage<>(i++);
  35. 	    	  }
  36. }

 

4. 应用主类

  1. package com.sande.streamhello;
  2.  
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5.  
  6.  
  7. @SpringBootApplication
  8. public class StreamHelloApplication {
  9.  
  10. 	public static void main(String[] args) {
  11. 		SpringApplication.run(StreamHelloApplication.class, args);
  12. 	}
  13. }

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

1. pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3. 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4. 	<modelVersion>4.0.0</modelVersion>
  5.  
  6. 	<groupId>com.sande</groupId>
  7. 	<artifactId>stream-hello</artifactId>
  8. 	<version>0.0.1-SNAPSHOT</version>
  9. 	<packaging>jar</packaging>
  10.  
  11. 	<name>stream-hello</name>
  12. 	<description>Demo project for Spring Boot</description>
  13.  
  14. 	<parent>
  15. 		<groupId>org.springframework.boot</groupId>
  16. 		<artifactId>spring-boot-starter-parent</artifactId>
  17. 		<version>2.0.4.RELEASE</version>
  18. 		<relativePath/> <!-- lookup parent from repository -->
  19. 	</parent>
  20.  
  21. 	<properties>
  22. 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  23. 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  24. 		<java.version>1.8</java.version>
  25. 		<spring-cloud.version>Finchley.SR1</spring-cloud.version>
  26. 	</properties>
  27.  
  28. 	<dependencies>
  29. 		<dependency>
  30. 			<groupId>org.springframework.boot</groupId>
  31. 			<artifactId>spring-boot-starter-web-services</artifactId>
  32. 		</dependency>
  33. 		<dependency>
  34. 			<groupId>org.springframework.cloud</groupId>
  35. 			<artifactId>spring-cloud-config-server</artifactId>
  36. 		</dependency>
  37. 		
  38.  
  39. 		<dependency>
  40. 			<groupId>org.springframework.boot</groupId>
  41. 			<artifactId>spring-boot-starter-test</artifactId>
  42. 			<scope>test</scope>
  43. 		</dependency>
  44. 		
  45. 		<dependency>
  46.             <groupId>org.springframework.cloud</groupId>
  47.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  48.             <version>2.0.1.RELEASE</version>
  49.         </dependency>
  50. 	</dependencies>
  51.  
  52. 	<dependencyManagement>
  53. 		<dependencies>
  54. 			<dependency>
  55. 				<groupId>org.springframework.cloud</groupId>
  56. 				<artifactId>spring-cloud-dependencies</artifactId>
  57. 				<version>${spring-cloud.version}</version>
  58. 				<type>pom</type>
  59. 				<scope>import</scope>
  60. 			</dependency>
  61. 		</dependencies>
  62. 	</dependencyManagement>
  63.  
  64. 	<build>
  65. 		<plugins>
  66. 			<plugin>
  67. 				<groupId>org.springframework.boot</groupId>
  68. 				<artifactId>spring-boot-maven-plugin</artifactId>
  69. 			</plugin>
  70. 		</plugins>
  71. 	</build>
  72.  
  73.  
  74. </project>

2. src/main/resources/application.properties

  1. spring.cloud.stream.bindings.input.group=Service-A
  2. spring.cloud.stream.bindings.input.destination=greetings
  3. server.port=8080
  4. spring.rabbitmq.host=localhost
  5. spring.rabbitmq.port=5672
  6. spring.rabbitmq.username=guest
  7. 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  消息消费组中的消费端节点。

3. 消息消费类

  1. package com.sande.streamhello.ramp;
  2.  
  3. import java.util.Date;
  4. import java.util.stream.Stream;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.cloud.stream.annotation.EnableBinding;
  8. import org.springframework.cloud.stream.annotation.StreamListener;
  9. import org.springframework.cloud.stream.messaging.Sink;
  10. import org.springframework.integration.annotation.ServiceActivator;
  11.  
  12. import com.sande.streamhello.StreamHelloApplication;
  13.  
  14. @EnableBinding(value= {Sink.class})
  15. public class SinkReceiver {
  16. 	
  17. 	private static Logger logger = LoggerFactory.getLogger(StreamHelloApplication.class);
  18. 	
  19. 	@StreamListener(Sink.INPUT)
  20. 	public void receive(String payload) {
  21. 		logger.info("Received:" + payload);
  22. 	}
  23. }

4、应用主类

  1. package com.sande.streamhello;
  2.  
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5.  
  6. @SpringBootApplication
  7. public class StreamHelloApplication {
  8.  
  9. 	public static void main(String[] args) {
  10. 		SpringApplication.run(StreamHelloApplication.class, args);
  11. 	}
  12. }

 

三、stream-hello3 节点（消息消费端）

1.pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3. 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4. 	<modelVersion>4.0.0</modelVersion>
  5.  
  6. 	<groupId>com.sande</groupId>
  7. 	<artifactId>stream-hello3</artifactId>
  8. 	<version>0.0.1-SNAPSHOT</version>
  9. 	<packaging>jar</packaging>
  10.  
  11. 	<name>stream-hello3</name>
  12. 	<description>Demo project for Spring Boot</description>
  13.  
  14. 	<parent>
  15. 		<groupId>org.springframework.boot</groupId>
  16. 		<artifactId>spring-boot-starter-parent</artifactId>
  17. 		<version>2.0.4.RELEASE</version>
  18. 		<relativePath/> <!-- lookup parent from repository -->
  19. 	</parent>
  20.  
  21. 	<properties>
  22. 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  23. 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  24. 		<java.version>1.8</java.version>
  25. 		<spring-cloud.version>Finchley.SR1</spring-cloud.version>
  26. 	</properties>
  27.  
  28. 	<dependencies>
  29. 		<dependency>
  30. 			<groupId>org.springframework.boot</groupId>
  31. 			<artifactId>spring-boot-starter-web-services</artifactId>
  32. 		</dependency>
  33. 		<dependency>
  34. 			<groupId>org.springframework.cloud</groupId>
  35. 			<artifactId>spring-cloud-config-server</artifactId>
  36. 		</dependency>
  37. 		
  38.  
  39. 		<dependency>
  40. 			<groupId>org.springframework.boot</groupId>
  41. 			<artifactId>spring-boot-starter-test</artifactId>
  42. 			<scope>test</scope>
  43. 		</dependency>
  44. 		
  45. 		<dependency>
  46.             <groupId>org.springframework.cloud</groupId>
  47.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  48.             <version>2.0.1.RELEASE</version>
  49.         </dependency>
  50. 	</dependencies>
  51.  
  52. 	<dependencyManagement>
  53. 		<dependencies>
  54. 			<dependency>
  55. 				<groupId>org.springframework.cloud</groupId>
  56. 				<artifactId>spring-cloud-dependencies</artifactId>
  57. 				<version>${spring-cloud.version}</version>
  58. 				<type>pom</type>
  59. 				<scope>import</scope>
  60. 			</dependency>
  61. 		</dependencies>
  62. 	</dependencyManagement>
  63.  
  64. 	<build>
  65. 		<plugins>
  66. 			<plugin>
  67. 				<groupId>org.springframework.boot</groupId>
  68. 				<artifactId>spring-boot-maven-plugin</artifactId>
  69. 			</plugin>
  70. 		</plugins>
  71. 	</build>
  72.  
  73.  
  74. </project>

2. src/main/resources/application.properties

  1. spring.cloud.stream.bindings.input.group=Service-A
  2. spring.cloud.stream.bindings.input.destination=greetings
  3. server.port=8082
  4. spring.rabbitmq.host=localhost
  5. spring.rabbitmq.port=5672
  6. spring.rabbitmq.username=guest
  7. 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  消息消费组中的消费端节点。

3. 消息消费类

  1. package com.sande.streamhello.ramp;
  2.  
  3. import java.util.stream.Stream;
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import org.springframework.cloud.stream.annotation.EnableBinding;
  7. import org.springframework.cloud.stream.annotation.StreamListener;
  8. import org.springframework.cloud.stream.messaging.Sink;
  9. import com.sande.streamhello.StreamHelloApplication;
  10.  
  11. @EnableBinding(value= {Sink.class})
  12. public class SinkReceiver {
  13. 	
  14. 	private static Logger logger = LoggerFactory.getLogger(StreamHelloApplication.class);
  15. 	
  16. 	@StreamListener(Sink.INPUT)
  17. 	public void receive(String payload) {
  18. 		logger.info("Received:" + payload);
  19. 	}
  20. }

4. 应用主类

  1. package com.sande.streamhello;
  2.  
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5.  
  6. @SpringBootApplication
  7. public class StreamHelloApplication {
  8.  
  9. 	public static void main(String[] args) {
  10. 		SpringApplication.run(StreamHelloApplication.class, args);
  11. 	}
  12. }

四、测试

1. 启动RabbitMQ

2. 启动节点 stream-hello 和 stream-hello3

3. 启动节点 stream-hello2

3. 我们看到 RabbitMQ 中已经创建了 greetings 交换器

 

4. 我们看到 RabbitMQ 中已经创建了greetings.Service-A 消息队列

5. 消费节点接收到的消息

5.1 stream-hello 节点接收的消息

5.2 stream-hello3 接收到的消息

通过两个消费端输出的消息我们看到，每条消息只会被其中一个消费节点接收。

 

五、配置消息

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

  1. spring.cloud.stream.bindings.output.destination=greetings
  2. spring.cloud.stream.bindings.output.producer.partition-count=2
  3. spring.cloud.stream.bindings.output.producer.partition-key-expression=1
  4. server.port=8081
  5. spring.rabbitmq.host=localhost
  6. spring.rabbitmq.port=5672
  7. spring.rabbitmq.username=guest
  8. 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

  1. spring.cloud.stream.bindings.input.group=Service-A
  2. spring.cloud.stream.bindings.input.destination=greetings
  3. spring.cloud.stream.bindings.input.consumer.partitioned=true
  4. spring.cloud.stream.instance-count=2
  5. spring.cloud.stream.instance-index=0
  6. server.port=8080
  7. spring.rabbitmq.host=localhost
  8. spring.rabbitmq.port=5672
  9. spring.rabbitmq.username=guest
  10. 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

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

4. 测试

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 接收到的消息的数字都是连续的。

4.3 当消息生产端 设置 spring.cloud.stream.bindings.output.producer.partition-key-expression=0 时，只有实例索引号为0的 stream-hello 节点可以接收到信息。

 

查看 RabbitMQ Management 可以看到，其实在 RabbitMQ 中，针对两个消费端分别创建了消息队列。

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号