热门标签
热门文章
- 12024年华中杯数学建模A题思路与论文助攻_华中杯24a题
- 2计量经济学学习笔记：多重共线性、异方差、自相关_多重共线性与自相关的关系
- 3【转】华为Java笔试题
- 4Gradio 案例——将 dicom 文件转为 nii文件_linux将dicm文件转为nii文件
- 5Cannot find class: com.mysql.cj.jdbc.Driver无法找到驱动文件_cannot find declaration to go to com.mysql.cj.jdbc
- 6编写高质量代码改善C++程序的150个建议_编写高质量c++程序
- 7C#/.NET面试问题总结 第一篇_c#实习生面试
- 8TIA Portal联合Process Simulate搞定SICAR虚拟调试实例步骤_tia+portal++sicar
- 9hfish蜜罐搭建与使用_hfish系统怎么停止ssh蜜罐
- 10【JavaEE 初阶（二）】线程安全问题
当前位置: article > 正文
Rabbitmq取消预取机制配置，配置手动确认后仍然java.lang.IllegalStateException: Channel closed； cannot ack/nack的问题_rabbitmq java.lang.illegalstateexception: connecti
作者：凡人多烦事01 | 2024-05-26 08:29:26
赞
踩
rabbitmq java.lang.illegalstateexception: connection returned a null channel
目录
背景
根据教程写了工作队列，然后发现2个消费者是公平消费的，即10条消息，2个队列，分别取到了5条消息进行消费，由于教程设置2个消费者分别线程休眠时间不一样，目的是让其消费速度不一致，模仿实际生产中消息消费过程时间有长有短。（一个消费慢，一个消费快，若公平消费的话，即平分当前消息量，快的消费者已经干完事在休息了，慢的消费者还有一大堆消息等着处理呢，明显不符合资源调配利用，应该能者多劳）
配置
因为想要实现“能者多劳”的功能，所以消息的应答方式一定要设置成手动应答manual，参数prefetch的含义表示RabbitMQ服务器节点上最多的没有确认的消息个数. 这个参数设置为0时,表示不对消息个数进行限制, RabbitMQ服务器会投递尽可能多的消息到consumer，这里我设置成1
- #yml配置
- spring:
- rabbitmq:
- host: 127.0.0.1
- port: 5672
- username: guest
- password: guest
- virtual-host: /
- listener:
- simple:
- prefetch: 1 # 每次只能获取一条，处理完成才能获取下一条
- acknowledge-mode: manual #一定改为要手动确认模式
报错
代码-生产者
- @Component
- public class PrefetchSender {
- @Autowired
- private RabbitTemplate rabbitTemplate;
-
- public void prefetch() {
- String exchange = "prefetch";
- String queue = "prefetch.queue";
- for (int i = 0; i < 10; i++) {
- rabbitTemplate.convertAndSend(exchange, queue, "消息" + i);
- }
- }
- }
- // 测试类方法
- @Test
- void prefetch() {
- prefetchSender.prefetch();
- }
代码-消费者
- import com.rabbitmq.client.Channel;
- import org.springframework.amqp.core.Message;
- import org.springframework.amqp.rabbit.annotation.Queue;
- import org.springframework.amqp.rabbit.annotation.RabbitListener;
- import org.springframework.stereotype.Component;
-
- import java.io.IOException;
-
- @Component
- public class PrefetchReceiver {
- // 消费者1，快
- @RabbitListener(queuesToDeclare = @Queue("prefetch.queue")) // queuesToDeclare 自动声明队列
- public void myListener1(Message message, Channel channel) throws IOException, InterruptedException {
- Thread.sleep(200);
- System.out.println("1收到的消息是：" + new String(message.getBody(), "utf-8"));
- channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
- }
-
- // 消费者2，慢
- @RabbitListener(queuesToDeclare = @Queue("prefetch.queue")) // queuesToDeclare 自动声明队列
- public void myListener2(Message message, Channel channel) throws IOException, InterruptedException {
- Thread.sleep(1000);
- System.out.println("2收到的消息是：" + new String(message.getBody(), "utf-8"));
- channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
- }
- }
思考
已经配置了手动确认（spring.rabbitmq.listener.simple.acknowledge-mode: manual），还有每次获取消息处理是1条，处理完消息才能拿下1条（spring.rabbitmq.listener.simple.prefetch: 1），为什么通道关闭，确认消息失败呢？
翻看一些帖子后，发现有些项目使用了消息转换json，我这个demo也使用了消息转json，目的是后台管理的界面可以看到发送的消息是json格式（未加密，但中文会乱码），但是消息接收的时候也需要进行json转换，需要在项目中加rabbitmq的配置类，连接工厂对象会在这个类里加工一下，所以发现问题所在了，配置类加工连接的时候漏了，需要我们"手动加上的"。
- import org.springframework.amqp.core.AcknowledgeMode;
- import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
- import org.springframework.amqp.rabbit.connection.ConnectionFactory;
- import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
- import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
- import org.springframework.context.annotation.Bean;
- import org.springframework.context.annotation.Configuration;
-
- @Configuration
- public class RabbitMQConfig {
- @Bean
- public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
- SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
- factory.setConnectionFactory(connectionFactory);
- factory.setMessageConverter(new Jackson2JsonMessageConverter()); // json转消息
- factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 手动确认（手动加上的）
- factory.setPrefetchCount(1); // 每次只能获取一条，处理完成才能获取下一条（手动加上的）
- return factory;
- }
- }
结果
总结
生产者项目配置了消息转json，那么，消费者项目是需要用配置类设置json转消息的，如果在yml或者properties配置文件修改了参数，不起作用的时候，需要检查配置类，手动加上变更的参数。（自己写demo的时候知道消息体可以转json就好了，还是不配置比较好。）
最后，谢谢大佬帖子给我启发，大家可以见原帖rabbitmq设置手动ack报错：Channel closed； cannot ack/nack_qq_42894258的博客-CSDN博客
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/凡人多烦事01/article/detail/625967
推荐阅读
相关标签
