SpringCloudStream+RabbitMQ使用中遇到的问题及解决
作者：星空_灯塔
问题一
一个交换机，多个队列，每个队列路由不同，向其中一个队列发送消息后，所有队列都能接收到消息
原因
没有绑定确定的路由
解决
添加生产者路由绑定
#配置方式一：A工程作为生产者生产消息，B工程作为消费者订阅消息
#生产者：A工程
spring:
cloud:
stream:
default-binder: rabbit
bindings:
A_stream:
binder: rabbit
destination: A-exchange-dev-4
group: group_A
rabbit:
bindings:
A_stream:
consumer:
exchangeType: topic
autoBindDlq: true
republishToDlq: true
bindingRoutingKey: group_A
producer:
bindingRoutingKey: group_A
routingKeyExpression: '''group_A'''
#消费者：B工程
spring:
cloud:
stream:
default-binder: rabbit
bindings:
A_stream:
binder: rabbit
destination: A-exchange-dev-4
group: group_A
rabbit:
bindings:
A_stream:
consumer:
exchangeType: topic
autoBindDlq: true
republishToDlq: true
bindingRoutingKey: group_A
producer:
bindingRoutingKey: group_A
routingKeyExpression: '''group_A'''问题二
两个不同交换机对应不同队列，队列组名称相同，生产者消费者绑定路由key相同，发送消息，目的交换机路由未收到消息，另一个却收到消息
原因
队列组相同
解决
修改队列组，使交换机路由组生产者消费者路由key不同，否则，不能使消息发送到目的路由
问题三
一个队列既是消费者又是生产者，如果生产者和消费者使用同一配置，容易出现查不到消费者的情况
解决
生产者和消费者使用不同通道
#配置方式二：同一工程中既是消费者又是生产者，生产者消费者分开配置
spring:
cloud:
stream:
default-binder: rabbit
bindings:
A_output_stream:
binder: rabbit
destination: A-exchange-dev-4
group: group_A
A_input_stream:
binder: rabbit
destination: A-exchange-dev-4
group: group_A
rabbit:
bindings:
A_output_stream:
producer:
bindingRoutingKey: group_A
routingKeyExpression: '''group_A'''
A_input_stream:
consumer:
exchangeType: topic
autoBindDlq: true
republishToDlq: true
bindingRoutingKey: group_A问题四
启动报错
如下：
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'exchange-dev-4-test001.product_sync' in vhost '/': received none but current is the value 'DLX' of type 'longstr', class-id=50, method-id=10)
......
org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: ......
Caused by: org.springframework.beans.factory.support.BeanDefinitionOverrideException: Invalid bean definition with name 'exchange-dev-4-test001.product_sync.errors.recoverer' defined in null: Cannot register bean definition [Root bean: class [org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer]; scope=; abstract=false; lazyInit=null; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=null; factoryMethodName=null; initMethodName=null; destroyMethodName=null] for bean 'exchange-dev-4-test001.product_sync.errors.recoverer': There is already [Root bean: class [org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer]; scope=; abstract=false; lazyInit=null; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=null; factoryMethodName=null; initMethodName=null; destroyMethodName=null] bound.
原因
bean 定义无效，已经有 bean 绑定。本次报错的原因是 bean 中对输入通道[@Input(ProductSyncQueueComponent .INPUT)]重复定义
在同一个工程中，[exchange-dev-4-test001.product_sync]既是消费者又是生产者，启动程序时报错，在 mq 队列查看 routingKey 时发现有两个，一个是自定义的[product_sync]，一个是默认的[#]，导致了生产者发送的消息消费者接收不到。
根据报错提示 bean 定义无效，检查自定义stream bean 时发现下列代码中生产者和消费者都定义成了输入通道，输入输出通道定义不正确。
@Component
public interface ProductSyncQueueComponent {
/**
* Input channel name.
*/
String INPUT = "product_sync_input_stream";
@Input(ProductSyncQueueComponent.INPUT)
SubscribableChannel input();
/**
* Output channel name.
*/
String OUTPUT = "product_sync_output_stream";
@Output(ProductSyncQueueComponent.OUTPUT)
SubscribableChannel output();
}解决
正确定义输入输出通道。(该解决方法仅使用于上述情况)
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。