springboot下使用rabbitMQ之开发配置方式(二)
springboot下使用rabbitMQ之传参及序列化(二)
消息参数传递在开发中也是个坑，不论使用内置的SimpleMessageConverter还是Jackson2JsonMessageConverter均无法让Consumer接收动态参数
一.序列化的问题
首先贴出具体代码以及测试用例:
- 消费者
@RabbitListener(queues = "text.queue")
@RabbitHandler(isDefault = true)
public void exec(@Payload Map dto, Message message, Channel channel){
// 注意，发送的消息类型必须是实现了Serializable接口的类型，消费者接口类型不能随便写!
LOG.info(RabbitMQCfgEnum.TEXT +"接收到消息:{}", dto);
// 设置手动确认才会需要执行此
//channel.basicAck(message.getMessageProperties().getDeliveryTag(),Boolean.TRUE);
}
- 生产者（测试用例）
private Connection buidConnection()throws Exception{
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();//MQ采用工厂模式来完成连接的创建
//2.在工厂对象中设置连接信息(ip,port,virtualhost,username,password)
factory.setHost("10.156.122.215");//设置MQ安装的服务器ip地址
factory.setPort(5672);//设置端口号
factory.setVirtualHost("vhost");//设置虚拟主机名称
//MQ通过用户来管理
factory.setUsername("shadow");//设置用户名称
factory.setPassword("shadow");//设置用户密码
//3.通过工厂对象获取连接
Connection connection = factory.newConnection();
return connection;
}
@Test
public void test02()throws Exception{
Connection connection = this.buidConnection();
Channel channel = connection.createChannel();
Map<String,Object> data = new HashMap<>(4);
data.put("ordeNo", SeqGenUtil.genSeq());
data.put("timestamp",System.currentTimeMillis());
String json = JacksonUtil.toJsonString(data);
channel.basicPublish(RabbitMQCfgEnum.TEXT.exchange, RabbitMQCfgEnum.TEXT.routingKey, null,json.getBytes(StandardCharsets.UTF_8));
//关闭连接
channel.close();
connection.close();
}
执行测试用例，它居然抛错了：
2023-07-17 14:10:24.037 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.rabbit.retry.RejectAndDontRequeueRecoverer:74 - Retries exhausted for message (Body:'[B@3ad45d58(byte[55])' MessageProperties [headers={}, contentLength=0, redelivered=false, receivedExchange=text_exchange, receivedRoutingKey=, deliveryTag=1, consumerTag=amq.ctag-IenKeUIVEFy1vBz1jQenVw, consumerQueue=text.queue])
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(java.util.Map,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@7839ec46]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy107.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [java.util.Map] for GenericMessage [payload=byte[55], headers={amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=c28511a8-1bbe-9d57-879c-c5763ba40129, amqp_consumerTag=amq.ctag-IenKeUIVEFy1vBz1jQenVw, amqp_lastInBatch=false, timestamp=1689574224032}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
2023-07-17 14:10:24.038 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.r.listener.ConditionalRejectingErrorHandler:170 - Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76)
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:78)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:157)
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:539)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:387)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy107.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: null
... 19 common frames omitted
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(java.util.Map,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@7839ec46]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
... 14 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [java.util.Map] for GenericMessage [payload=byte[55], headers={amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=c28511a8-1bbe-9d57-879c-c5763ba40129, amqp_consumerTag=amq.ctag-IenKeUIVEFy1vBz1jQenVw, amqp_lastInBatch=false, timestamp=1689574224032}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
看吧json字符串是无法序列化为Map😂
试试用dto来接收这个json字符串看:
- 消费者
@RabbitListener(queues = "text.queue")
@RabbitHandler(isDefault = true)
public void exec(@Payload MQMessageDTO dto, Message message, Channel channel){
// 注意，发送的消息类型必须是实现了Serializable接口的类型，消费者接口类型不能随便写!
LOG.info(RabbitMQCfgEnum.TEXT +"接收到消息:{}", dto);
// 设置手动确认才会需要执行此
//channel.basicAck(message.getMessageProperties().getDeliveryTag(),Boolean.TRUE);
}
- 消费者接收对象
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import java.io.Serializable;
@JsonIgnoreProperties(ignoreUnknown = true)
public class MQMessageDTO implements Serializable {
/**
* 序列化标识
*/
private static final long serialVersionUID = 1L;
private String ordeNo;
private Long timestamp;
public String getOrdeNo() {
return ordeNo;
}
public void setOrdeNo(String ordeNo) {
this.ordeNo = ordeNo;
}
public Long getTimestamp() {
return timestamp;
}
public void setTimestamp(Long timestamp) {
this.timestamp = timestamp;
}
@Override
public String toString() {
return "MQMessageDTO{" +
"ordeNo='" + ordeNo + '\'' +
", timestamp=" + timestamp +
'}';
}
}
再执行下测试用例看看:
2023-07-17 14:15:14.484 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.rabbit.retry.RejectAndDontRequeueRecoverer:74 - Retries exhausted for message (Body:'[B@20c64033(byte[55])' MessageProperties [headers={}, contentLength=0, redelivered=false, receivedExchange=text_exchange, receivedRoutingKey=, deliveryTag=1, consumerTag=amq.ctag-a17LF_eoKfyC9ynPPdJmfw, consumerQueue=text.queue])
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(com.mee.api.common.dto.MQMessageDTO,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@616a95dd]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy107.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.mee.api.common.dto.MQMessageDTO] for GenericMessage [payload=byte[55], headers={amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=42e348e4-aca4-fc9e-46f3-b52c83295a0c, amqp_consumerTag=amq.ctag-a17LF_eoKfyC9ynPPdJmfw, amqp_lastInBatch=false, timestamp=1689574514480}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
2023-07-17 14:15:14.485 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.r.listener.ConditionalRejectingErrorHandler:170 - Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76)
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:78)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:157)
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:539)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:387)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy107.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: null
... 19 common frames omitted
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(com.mee.api.common.dto.MQMessageDTO,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@616a95dd]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
... 14 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.mee.api.common.dto.MQMessageDTO] for GenericMessage [payload=byte[55], headers={amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=42e348e4-aca4-fc9e-46f3-b52c83295a0c, amqp_consumerTag=amq.ctag-a17LF_eoKfyC9ynPPdJmfw, amqp_lastInBatch=false, timestamp=1689574514480}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
还是一样的抛错😂
没办法，我们得用先前能走通的调试下Consumer的Message这个参数看:
看到了有个貌似是参数类型的东东出现在了inferredArgumentType这个参数位置,对头～，可以试着将类型放进去不就好了嘛，试试看~
- 消费者
@RabbitListener(queues = "default.queue")
@RabbitHandler(isDefault = true)
public void exec(@Payload MQMessageDTO dto, Message message, Channel channel){
// 注意，发送的消息类型必须是实现了Serializable接口的类型，消费者接口类型不能随便写!
LOG.info(RabbitMQCfgEnum.DEFAULT +"接收到消息:{}", dto);
}
- 测试用例
@Test
public void test03()throws Exception{
Connection connection = this.buidConnection();
Channel channel = connection.createChannel();
MQMessageDTO data = new MQMessageDTO();
data.setOrdeNo(SeqGenUtil.genSeq());
data.setTimestamp(System.currentTimeMillis());
MessageProperties messageProperties = new MessageProperties();
messageProperties.setInferredArgumentType(MQMessageDTO.class);
messageProperties.setContentType("application/x-java-serialized-object");
Message message = new SimpleMessageConverter().toMessage(data, messageProperties);
channel.basicPublish(RabbitMQCfgEnum.TEXT.exchange, RabbitMQCfgEnum.TEXT.routingKey, null,message.getBody());
//关闭连接
channel.close();
connection.close();
}
试着运行了下测试用例，发现还是抛错😂
2023-07-17 15:31:26.051 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.rabbit.retry.RejectAndDontRequeueRecoverer:74 - Retries exhausted for message (Body:'[B@2ac0f60d(byte[214])' MessageProperties [headers={}, contentLength=0, redelivered=false, receivedExchange=text_exchange, receivedRoutingKey=, deliveryTag=1, consumerTag=amq.ctag-YZyegHnG7Y_t5bio0BmSvQ, consumerQueue=text.queue])
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(com.mee.api.common.dto.MQMessageDTO,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@3388051d]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy108.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.mee.api.common.dto.MQMessageDTO] for GenericMessage [payload=byte[214], headers={amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=51ebd117-d95b-463a-f43b-7feff582b194, amqp_consumerTag=amq.ctag-YZyegHnG7Y_t5bio0BmSvQ, amqp_lastInBatch=false, timestamp=1689579086047}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
2023-07-17 15:31:26.052 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.r.listener.ConditionalRejectingErrorHandler:170 - Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76)
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:78)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:157)
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:539)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:387)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy108.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: null
... 19 common frames omitted
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(com.mee.api.common.dto.MQMessageDTO,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@3388051d]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
... 14 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.mee.api.common.dto.MQMessageDTO] for GenericMessage [payload=byte[214], headers={amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=51ebd117-d95b-463a-f43b-7feff582b194, amqp_consumerTag=amq.ctag-YZyegHnG7Y_t5bio0BmSvQ, amqp_lastInBatch=false, timestamp=1689579086047}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
可能思路还是存在问题，这样吧，先用RabbitTemplate推一个消息看
主要代码如下:
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDefaultDirect() {
MQMessageDTO data = new MQMessageDTO();
data.setOrdeNo(SeqGenUtil.genSeq());
data.setTimestamp(System.currentTimeMillis());
// 注意，发送的消息必须是实现了Serializable类型的消息，接收的时候也应该是这个类型,不可乱写
rabbitTemplate.convertAndSend(RabbitMQCfgEnum.DEFAULT.exchange, RabbitMQCfgEnum.DEFAULT.routingKey,data);
LOG.info("已发送::{}",data );
}
能收到消息:
2023-07-17 15:36:17.354 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1] -> INFO c.m.a.common.service.impl.MQConsumerDefaultHandler:49 -
DEFAULT接收到消息:MQMessageDTO{ordeNo='2307171536109000', timestamp=1689579370802}
花了幾分鐘，試著在RabbitTemplate下找到了这段代码：
protected void sendToRabbit(Channel channel, String exchange, String routingKey, boolean mandatory,
Message message) throws IOException {
BasicProperties convertedMessageProperties = this.messagePropertiesConverter
.fromMessageProperties(message.getMessageProperties(), this.encoding);
channel.basicPublish(exchange, routingKey, mandatory, convertedMessageProperties, message.getBody());
}
好家伙，原来需要传一个 BasicProperties 对象，copy之~
- 测试用例
@Test
public void test03()throws Exception{
Connection connection = this.buidConnection();
//mq提供Channel来将处理消息
//创建Channel
Channel channel = connection.createChannel();
MQMessageDTO data = new MQMessageDTO();
data.setOrdeNo(SeqGenUtil.genSeq());
data.setTimestamp(System.currentTimeMillis());
Message message = new SimpleMessageConverter().toMessage(data, new MessageProperties());
AMQP.BasicProperties prop = new DefaultMessagePropertiesConverter().fromMessageProperties(message.getMessageProperties(), "UTF-8");
channel.basicPublish(RabbitMQCfgEnum.TEXT.exchange, RabbitMQCfgEnum.TEXT.routingKey, prop,message.getBody());
//关闭连接
channel.close();
connection.close();
}
- 执行结果
2023-07-17 15:41:20.880 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> INFO c.m.api.common.service.impl.MQConsumerTextHandler:43 -
TEXT接收到消息:MQMessageDTO{ordeNo='2307171541179000', timestamp=1689579677767}
基本结论:
在springboot框架下mq消费者接收生产者的消息内容，如果想统一封装则只能用字符串的形式！这点儿很重要，因为发送方配置中指定了发送消息的内容类型~
但是，如果生产者与消费者不是在一个应用，能不能用同一个类型接收呢，这是个好问题，试试看~
二.模拟发送方与接收方参数测试
为了真实模拟，我将上面使用过的这个消息体MQMessageDTO copy一份到其他包里面,生产者与消费者使用不同包下的MQMessageDTO：
最终的代码是这样子:
- 消费者
@RabbitListener(queues = "text.queue")
@RabbitHandler(isDefault = true)
public void exec(@Payload com.mee.api.common.dto.MQMessageDTO dto, Message message, Channel channel){
// 注意，发送的消息类型必须是实现了Serializable接口的类型，消费者接口类型不能随便写!
LOG.info(RabbitMQCfgEnum.TEXT +"接收到消息:{}", dto);
// 设置手动确认才会需要执行此
//channel.basicAck(message.getMessageProperties().getDeliveryTag(),Boolean.TRUE);
}
- 生产者(测试用例)
@Test
public void test03()throws Exception{
Connection connection = this.buidConnection();
//mq提供Channel来将处理消息
//创建Channel
Channel channel = connection.createChannel();
com.mee.api.common.dto.tmp.MQMessageDTO data = new com.mee.api.common.dto.tmp.MQMessageDTO();
data.setOrdeNo(SeqGenUtil.genSeq());
data.setTimestamp(System.currentTimeMillis());
Message message = new SimpleMessageConverter().toMessage(data, new MessageProperties());
AMQP.BasicProperties prop = new DefaultMessagePropertiesConverter().fromMessageProperties(message.getMessageProperties(), "UTF-8");
channel.basicPublish(RabbitMQCfgEnum.TEXT.exchange, RabbitMQCfgEnum.TEXT.routingKey, prop,message.getBody());
//关闭连接
channel.close();
connection.close();
}
- 测试结果
2023-07-17 15:55:03.410 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.rabbit.retry.RejectAndDontRequeueRecoverer:74 - Retries exhausted for message (Body:'[serialized object]' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=text_exchange, receivedRoutingKey=, deliveryTag=1, consumerTag=amq.ctag-aS3e5aUBnUMAoQQlxU7QQQ, consumerQueue=text.queue])
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(com.mee.api.common.dto.MQMessageDTO,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@292b4ff]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy108.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [com.mee.api.common.dto.tmp.MQMessageDTO] to [com.mee.api.common.dto.MQMessageDTO] for GenericMessage [payload=MQMessageDTO{ordeNo='2307171555019000', timestamp=1689580501335}, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=31472e0d-b0a5-676e-2899-e1542410937f, amqp_consumerTag=amq.ctag-aS3e5aUBnUMAoQQlxU7QQQ, amqp_lastInBatch=false, contentType=application/x-java-serialized-object, timestamp=1689580503404}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
2023-07-17 15:55:03.411 -> [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#1-1] -> WARN o.s.a.r.listener.ConditionalRejectingErrorHandler:170 - Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76)
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:78)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:157)
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:539)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:387)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
at org.springframework.amqp.rabbit.listener.$Proxy108.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: null
... 19 common frames omitted
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(com.mee.api.common.dto.MQMessageDTO,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@292b4ff]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
... 14 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [com.mee.api.common.dto.tmp.MQMessageDTO] to [com.mee.api.common.dto.MQMessageDTO] for GenericMessage [payload=MQMessageDTO{ordeNo='2307171555019000', timestamp=1689580501335}, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=text_exchange, amqp_deliveryTag=1, amqp_consumerQueue=text.queue, amqp_redelivered=false, id=31472e0d-b0a5-676e-2899-e1542410937f, amqp_consumerTag=amq.ctag-aS3e5aUBnUMAoQQlxU7QQQ, amqp_lastInBatch=false, contentType=application/x-java-serialized-object, timestamp=1689580503404}]
at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
... 27 common frames omitted
这上面有一句话很重要：
Cannot convert from [com.mee.api.common.dto.tmp.MQMessageDTO] to [com.mee.api.common.dto.MQMessageDTO]
所以在不同的应用下使用mq ，需要用相同包下的相同类传递参数。
总结：springboot下mq序列化的方式有三
-
1.使用默认的
SimpleMessageConverter实现序列化 -
2.配置使用
Jackson2JsonMessageConverter(jackson),配置如下
@Bean
public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory,ObjectMapper objectMapper){
SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory);
simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter(objectMapper));
return simpleRabbitListenerContainerFactory;
}
@Bean
public RabbitTemplate jackRabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
return rabbitTemplate;
}
@Bean
public MessageConverter jsonConvert(ObjectMapper objectMapper){
return new Jackson2JsonMessageConverter(objectMapper);
}
- 3.统一
String传递消息内容，接收时用如下方式还原为String
String msgStr = new String(bt,StandardCharsets.UTF_8);
如需要转换为特定类型请自行使用fastjson,jackson,gson等等方式将字符串序列化为指定对象即可～
funnyzpc@gmail.com