springboot下使用rabbitMQ之开发配置方式(二)

springboot下使用rabbitMQ之传参及序列化(二)

消息参数传递在开发中也是个坑，不论使用内置的SimpleMessageConverter还是Jackson2JsonMessageConverter均无法让Consumer接收动态参数
一.序列化的问题

首先贴出具体代码以及测试用例:

• 消费者
  

1.     @RabbitListener(queues = "text.queue")
2.     @RabbitHandler(isDefault = true)
3.     public void exec(@Payload Map dto, Message message, Channel channel){
4.         // 注意，发送的消息类型必须是实现了Serializable接口的类型，消费者接口类型不能随便写!
5.         LOG.info(RabbitMQCfgEnum.TEXT +"接收到消息:{}", dto);
6.         // 设置手动确认才会需要执行此
7.         //channel.basicAck(message.getMessageProperties().getDeliveryTag(),Boolean.TRUE);
8.     }

复制代码

• 生产者（测试用例）
  

1.     private Connection buidConnection()throws Exception{
2.         //1.创建连接工厂
3.         ConnectionFactory factory = new ConnectionFactory();//MQ采用工厂模式来完成连接的创建
4.         //2.在工厂对象中设置连接信息(ip,port,virtualhost,username,password)
5.         factory.setHost("10.156.122.215");//设置MQ安装的服务器ip地址
6.         factory.setPort(5672);//设置端口号
7.         factory.setVirtualHost("vhost");//设置虚拟主机名称
8.         //MQ通过用户来管理
9.         factory.setUsername("shadow");//设置用户名称
10.         factory.setPassword("shadow");//设置用户密码
11.         //3.通过工厂对象获取连接
12.         Connection connection = factory.newConnection();
13.         return connection;
14.     }
15.    
16.     @Test
17.     public void test02()throws Exception{
18.         Connection connection = this.buidConnection();
19.         Channel channel = connection.createChannel();
20.         Map<String,Object> data = new HashMap<>(4);
21.         data.put("ordeNo", SeqGenUtil.genSeq());
22.         data.put("timestamp",System.currentTimeMillis());
23.         String json = JacksonUtil.toJsonString(data);
24.         channel.basicPublish(RabbitMQCfgEnum.TEXT.exchange, RabbitMQCfgEnum.TEXT.routingKey, null,json.getBytes(StandardCharsets.UTF_8));
25.         //关闭连接
26.         channel.close();
27.         connection.close();
28.     }

复制代码

执行测试用例，它居然抛错了：

1. 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])
2. org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
3. Endpoint handler details:
4. Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(java.util.Map,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
5. Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@7839ec46]
6.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
7.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
8.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
9.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
10.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
11.         at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
12.         at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
13.         at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
14.         at java.base/java.lang.reflect.Method.invoke(Method.java:568)
15.         at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
16.         at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
17.         at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
18.         at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
19.         at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
20.         at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
21.         at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
22.         at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
23.         at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
24.         at org.springframework.amqp.rabbit.listener.$Proxy107.invokeListener(Unknown Source)
25.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
26.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
27.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
28.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
29.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
30.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
31.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
32.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
33.         at java.base/java.lang.Thread.run(Thread.java:833)
34. 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}]
35.         at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
36.         at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
37.         at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
38.         at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
39.         at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
40.         at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
41.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
42.         ... 27 common frames omitted
43. 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.
44. org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
45.         at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76)
46.         at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:78)
47.         at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:157)
48.         at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:539)
49.         at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:387)
50.         at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
51.         at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:122)
52.         at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
53.         at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:220)
54.         at org.springframework.amqp.rabbit.listener.$Proxy107.invokeListener(Unknown Source)
55.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1581)
56.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1572)
57.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1516)
58.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1001)
59.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:948)
60.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:86)
61.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1326)
62.         at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1232)
63.         at java.base/java.lang.Thread.run(Thread.java:833)
64. Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: null
65.         ... 19 common frames omitted
66. Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
67. Endpoint handler details:
68. Method [public void com.mee.api.common.service.impl.MQConsumerTextHandler.exec(java.util.Map,org.springframework.amqp.core.Message,com.rabbitmq.client.Channel)]
69. Bean [com.mee.api.common.service.impl.MQConsumerTextHandler@7839ec46]
70.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:267)
71.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:209)
72.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:148)
73.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1674)
74.         at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1593)
75.         at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
76.         at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
77.         at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
78.         at java.base/java.lang.reflect.Method.invoke(Method.java:568)
79.         at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
80.         at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
81.         at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
82.         at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:97)
83.         at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
84.         ... 14 common frames omitted
85. 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}]
86.         at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
87.         at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
88.         at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
89.         at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
90.         at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
91.         at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:77)
92.         at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
93.         ... 27 common frames omitted

复制代码

<strong>看吧json字符串是无法序列化为Map
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！