spring.cloud.stream3.0整合rabbitmq
本文简单介绍 spring cloud stream 3.x较之前版本有很大的不同,废除了@Input、@Output、@EnableBinding、@StreamListener等注解 以下内容包括生产者消费及单元测试
1、引入gradle依赖
ext { set(springCloudVersion, "2020.0.1") } dependencies { implementation org.springframework.cloud:spring-cloud-stream-binder-rabbit } dependencyManagement { imports { mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}" } }
2、配置文件
rabbitmq配置及消费者配置,生产者可不配置
spring: rabbitmq: host: rabbitmq1.com,rabbitmq2.com cloud: stream: defaultBinder: rabbitmq binders: rabbitmq: type: rabbitmq bindings: handleTest-in-0: # 生产者发送的 destination: test-des # 消费组 group: test_group rabbit: bindings: handleTest-in-0: consumer: autoBindDlq: true # 使用自定义统一的死信队列 deadLetterQueueName: system.global-dlq.system # 单个可不指定,多个使用`;`分号隔开 function: definition: handleTest
binding命名规则 上面代码中的binding名称都是自动生成的,命名规则是: input - <方法名> + -in- + output - <方法名> + -out- + Supplier为output,Consumer为input。除非方法参数有多个input或output,否则index为0。
3、生产者
@Component @AllArgsConstructor public class TestPublisher { private final StreamBridge streamBridge; /** * 发送消息 */ public void publish(String str) { streamBridge.send("test-des", str); } }
4、消费者
@Component public class TestListener { /** * 处理消息 */ @Bean public Consumer<String> handleTest() { return str-> { System.out.println(str); }; } }
5、单元测试
引入stream单元测试gradle依赖
testImplementation("org.springframework.cloud:spring-cloud-stream") { artifact { name = "spring-cloud-stream" extension = "jar" type = "test-jar" classifier = "test-binder" } }
如果之前添加了spring-cloud-stream-test-support依赖,需要先删除,避免冲突
@SpringBootTest @Import({ TestChannelBinderConfiguration.class}) //不会真实发送 public class AuthChangeListenerTests { @Resource private InputDestination input; @Resource private OutputDestination output; /** * 测试发送mq */ @Test public void testInput() { //发送一条消息到交换机test-des input.send(MessageBuilder.withPayload("test").build(), "test-des"); //可以立即对消息处理结果进行判断,上一步会处理完再往下运行 assert ... } /** * 测试接收mq */ @Test public void testOutput() { // 业务代码写在此处 // 判断mq是否发送到交换机test-des Message<byte[]> outputMessage = output.receive(0, "test-des"); assertThat(outputMessage.getPayload()).isEqualTo("test".getBytes()); } }
注意:若测试发送消息,本项目也是消费者,发送的消息会被项目优先消费,output.receive无法再收取消息