SpringBoot整合activeMQ消息队列手动签收（Session.CLIENT

今天在家隔离办公，不太忙，然后就琢磨起来消息队列activeMQ的消息事务来解决分布式事务，但是奈何在SpringBoot整合activeMQ时，其消费者手动签收消息时出现了问题——>当activeMQ设置为手动签收(Session.CLIENT_ACKNOWLEDGE)时，我明明在消费消息时没有执行mapMessage.acknowledge()方法，但是却将消息消费成功了，可郁闷了，哪也就是这个手动签收完全没有生效，郁闷死了，搞了两个小时完全没搞懂，然后网上也找不到类似的踩坑记录，后来给解决了这个坑，那么为了让咱们不再踩坑，这里我来给大家解决下，废话少说，老规矩，来扔干货。
大家能搜到我这篇文章，想必大家肯定已经对activeMQ已经有了大部分了解了，我就不给大家从头一点一点的讲解其知识了，直接说这个解决问题。
为了节省大家时间，大家可以看这段代码直接解决问题，如果能解决就不用看下面的废话，下面一堆废话，实际有用的也就这一点——> 就是在config配置类中设置factory.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);更改为factory.setSessionAcknowledgeMode(4);即可解决手动签收失败问题。

上面哪一段大家如果没有看懂也没关系，那么请接着往下读详细内容

在SpringBoot整合activeMQ之前，我创建了最基本的那种生产者和消费者，在这种消费者类中我设置的其消息签收模式为手动签收，这种我测试的是成功的，也就是只有在执行ackno wledge()方法之后，其消息才是真正签收成功，否则消息签收失败，消息仍然在消息队列中，其它消费者仍然可以在该消息队列中拿到其消息进行消费。代码如下：

public static void main(String[] args) {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");Connection connection = null;try {connection = activeMQConnectionFactory.createConnection();connection.start();/*** 第一个参数：是否开启事务；第二个参数：签收的几种方式* 自动签收（Session.AUTO_ACKNOWLEDGE）：该方式是默认的。该种方式，无需我们程序做任何操作，框架会帮我们自动签收收到的消息。* 手动签收（Session.CLIENT_ACKNOWLEDGE）：手动签收。该种方式，需要我们手动调用Message.acknowledge()，来签收消息。如果不签收消息。该消息会被我们反复消费，直到被签收* 允许重复消息（Session.DUPS_OK_ACKNOWLEDGE）：多线程或多个 消费者同时消费到一个消息，因为先成功不安全，可能会重复消费。该种方式很少使用到。* 事务下的签收（SESSION.SESSION_TRANSACTED）：开始事务的情况下，可以使用该方式。该种方式很少使用到。*///Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//Session session = connection.createSession(true, Session.SESSION_TRANSACTED);Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);Queue queue = session.createQueue("test01");MessageConsumer consumer = session.createConsumer(queue);consumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {String text = null;try {text = ((TextMessage) message).getText();System.out.println(text);message.acknowledge();} catch (JMSException e) {e.printStackTrace();}}}});} catch (JMSException e) {e.printStackTrace();}}

上面这个消费者就是让大家看下其知识点，注意点在以下两个地方：
上面中给大家展示这么多，就是为了告诉大家如果想要消费者手动签收消费，那么首先要将session设置为手动签收模式(Session.CLIENT_ACKNOWLEDGE)，然后在消费消息之后，执行acknowledge()方法，则该消息会签收成功。
在上面这种方式中已经手动签收成功，但是难就难在了在整合SpringBoot的时候却失败了，失败了，郁闷，下面说解决方案：

解决方案：

下面我依次给大家activeMQ的配置类、activeMQ工具类和消费端代码，如下所示：
配置类config如下所示：

/*** ActiveMQ 配置类* @ClassName: ActiveMQConfig* @Author: ygl* @Date: 2022/3/30 16:14* @Version: 1.0*/
@Configuration
public class ActiveMQConfig {@Value("${spring.activemq.broker-url:disabled}")String brokerURL;@Value("${activemq.listener.enable:disabled}")String listenerEnable;@Beanpublic ActiveMQUtil getActiveMQUtil() throws JMSException {if (brokerURL.equals("disabled")) {return null;}ActiveMQUtil activeMQUtil = new ActiveMQUtil();activeMQUtil.init(brokerURL);return activeMQUtil;}//定义一个消息监听器连接工厂，这里定义的是点对点模式的监听器连接工厂@Bean(name = "jmsQueueListener")public DefaultJmsListenerContainerFactory jmsQueueListenerContainerFactory(ActiveMQConnectionFactory activeMQConnectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();if (!listenerEnable.equals("true")) {return null;}factory.setConnectionFactory(activeMQConnectionFactory);//设置并发数factory.setConcurrency("5");//重连间隔时间factory.setRecoveryInterval(5000L);//设置为点对点(p2p)模式factory.setPubSubDomain(false);//作为生产者如果需要支持事务，则需要配置SessionTransacted为truefactory.setSessionTransacted(false);//消息的应答方式，需要手动确认，此时SessionTransacted必须被设置为false，且为Session.CLIENT_ACKNOWLEDGE模式//Session.AUTO_ACKNOWLEDGE  消息自动签收//Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收  注意在SpringBoot里面手动确认是失效的，要设置为 4//Session.DUPS_OK_ACKNOWLEDGE 不必必须签收，消息可能会重复发送//factory.setSessionAcknowledgeMode(4);factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);return factory;}@Bean(name = "topicListenerContainerFactory")public JmsListenerContainerFactory topicListenerContainerFactory(ConnectionFactory connectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();if (!listenerEnable.equals("true")) {return null;}factory.setConnectionFactory(connectionFactory);factory.setPubSubDomain(true);return factory;}@Beanpublic ActiveMQConnectionFactory activeMQConnectionFactory() {/*if((url==null||url.equals(""))&&!brokerURL.equals("disabled")){url=brokerURL;}*/ActiveMQConnectionFactory activeMQConnectionFactory =new ActiveMQConnectionFactory(brokerURL);return activeMQConnectionFactory;}
}

spring.activemq.broker-url和activemq.listener.enable在yml配置中如下图所示：

activeMQUtil工具类如下所示：

public class ActiveMQUtil {PooledConnectionFactory pooledConnectionFactory = null;public ConnectionFactory init(String brokerUrl) {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);//加入连接池pooledConnectionFactory = new PooledConnectionFactory(factory);//出现异常时重新连接pooledConnectionFactory.setReconnectOnException(true);//pooledConnectionFactory.setMaxConnections(5);pooledConnectionFactory.setExpiryTimeout(10000);return pooledConnectionFactory;}public ConnectionFactory getConnectionFactory() {return pooledConnectionFactory;}
}

消费端代码如下所示：

/*** 消息队列 监听* @ClassName: MQListener* @Author: ygl* @Date: 2022/3/30 21:44* @Version: 1.0**/
@Component
public class MQListener {@JmsListener(destination = "payment_testMQ_producer", containerFactory = "jmsQueueListener")
public void testMQConsumerThree(MapMessage mapMessage, Session session) {String name = null;String age = null;try {name = mapMessage.getString("name");age = mapMessage.getString("age");TestActiveMQDTO testActiveMQDTO = new TestActiveMQDTO();testActiveMQDTO.setName(name);testActiveMQDTO.setAge(Integer.parseInt(age));System.out.println("我是Three——>姓名：" + name + ";年龄：" + age);//session.rollback();mapMessage.acknowledge();} catch (JMSException e) {e.printStackTrace();}}
}

好，上面就是我的所有代码，其实只有三个点比较重要，我来给大家在下面的图片中标注出来：

上面这两张照片截图中已经详细的介绍了activeMQ的config配置和消费者监听的类，看着上面完全没问题，对，你看的是不是也没问题，但是我在消费时不执行mapMessage.acknowledge();方法去手动签收，但是该消息队列仍然签收成功，可郁闷可郁闷，为什么会呢？
后面网上找了资源，看下些源码发现源码中有这样一段代码：

上面图中的源码很清楚的介绍了为什么自动执行，我也在图中标注的比较清楚，这里我也不过多解释，请仔细看上面图中源码。
解决方案：
     就是在config配置类中设置 factory.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);更改为factory.setSessionAcknowledgeMode(4); 即可解决手动签收失败问题。
     好了，这就是我今天activeMQ消息手动签收遇到的问题，哪里有需要改正的和大家不懂的请帮忙给我指出，我将改正或者在评论里给大家再讲解下。
     大家如果想详细的了解activeMQ，请移步至这篇文章——>activeMQ基础学习和SpringBoot整合activeMQ案例；这篇文章中有详细介绍。
     大家记得点赞收藏加评论哦，明天上午继续加油，争取出来activeMQ的详细讲解。明天见，拜拜。

SpringBoot整合activeMQ消息队列手动签收（Session.CLIENT_ACKNOWLEDGE）为什么失效啊？相关推荐

springboot整合redis消息队列
前言消息队列作为一种常用的异步通信解决方案,而redis是一款高性能的nosql产品,今天就给大家介绍一下,如何使用redis实现消息队列,并整合到springboot. 两个消息模型 1. 队列模 ...
SpringBoot集成ActiveMq消息队列实现即时和延迟处理
原文链接:https://blog.csdn.net/My_harbor/article/details/81328727 一.安装ActiveMq 具体安装步骤:自己谷歌去二.新建springbo ...
SpringBoot整合RabbitMQ消息队列
RabbitMQ 一.RabbitMQ介绍 1.1 现存问题服务调用:两个服务调用时,我们可以通过传统的HTTP方式,让服务A直接去调用服务B的接口,但是这种方式是同步的方式,虽然可以采用Sprin ...
SpringBoot整合MQ消息队列
SpringBoot整合MQ 借鉴的文章 1.什么是MQ 2.消息队列可以做什么 3.下载安装MQ 4.SpringBoot整合MQ的步骤借鉴的文章 https://www.jianshu.com/ ...
activeMQ基础学习和SpringBoot整合activeMQ案例
昨天仔细研究了activeMQ消息队列,也遇到了些坑,昨天晚上也写了篇文章记录坑的内容,其实上篇文章(SpringBoot整合activeMQ消息队列手动签收(Session.CLIENT_ACKNO ...
SpringBoot整合RabbitMQ 消息可靠投递、手动ack、延迟队列、死信队列、消息幂等性保障、消息积压
1.消息可靠投递在使用 RabbitMQ 的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景.RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式. confirm 确认模式 ...
springboot：整合redis消息队列
整合redis消息队列项目依赖 <dependency><groupId>org.springframework.bo ...
springboot整合ActiveMQ(点对点和发布订阅)
springboot整合ActiveMQ(点对点和发布订阅) ActiveMQ是什么,为什么使用MQ 是基于 Java 中的 JMS 消息服务规范实现的一个消息中间件. 1.系统解耦采用中间件之后, ...
SpringBoot SpringBoot 开发实用篇 5 整合第三方技术 5.21 SpringBoot 整合 ActiveMQ
SpringBoot [黑马程序员SpringBoot2全套视频教程,springboot零基础到项目实战(spring boot2完整版)] SpringBoot 开发实用篇文章目录 Spring ...

SpringBoot整合activeMQ消息队列手动签收（Session.CLIENT_ACKNOWLEDGE）为什么失效啊？

解决方案：

SpringBoot整合activeMQ消息队列手动签收（Session.CLIENT_ACKNOWLEDGE）为什么失效啊？相关推荐

最新文章

热门文章