spring.cloud.stream3.0整合rabbitmq

68 阅读 0 评论 45 点赞

我是靠谱客的博主背后月亮，最近开发中收集的这篇文章主要介绍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());
}
}