ActiveMQ消息签收机制:
客户端成功接收一条消息的标志是一条消息被签收,成功应答。
消息的签收请求分为两种:
1.带事务的session
如果session带有事务,并且事务成功提交,则消息被自动签收。如果事务回滚,则消息会被再次传送。
2.不带事务的session
不带事务的session的签收方式,取决于session的配置
ActiveMQ支持以下三种模式:
Seesion.AUTO_ACKNOWLEDGE:消息自动签收;
Session.CLIENT_ACKNOWLEDGE:客户端调用acknowledge方法手动签收
textMessage.acknowledge();//手动签收
Session.DUPS_OK_ACKNOWLEDGE:不是必须签收,消息可能会重复发送。在第二次重新传送消息的时候,消息只有在被确认之后,才认为已经被成功地消费了。消息的成功消费通常包含三个阶段:客户端接收消息客户处理消息和消息被确认。在事务性会话中,当一个事务提交的时候,确认自动发生。在非事务性会话中,消息何时被确认取决于创建会话时的应答模式。
带事务session的案例
生产者
必须在生产完数据之后手动提交session
package com.wn.ddd; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Producter { public static void main(String[] args) throws JMSException { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); // JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); //启动连接 connection.start(); // Session: 一个发送或接收消息的线程 false:代表不带事务的session AUTO_ACKNOWLEDGE:代表自动签收 Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // Destination :消息的目的地;消息发送给谁. // 获取session注意参数值my-queue是Query的名字 Queue queue = session.createQueue("my-queue"); // MessageProducer:创建消息生产者 MessageProducer producer = session.createProducer(queue); // 设置不持久化 PERSISTENT:代表持久化 NON_PERSISTENT:代表不持久化 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 发送消息 for (int i = 1; i <= 5; i++) { sendMsg(session, producer, i); } System.out.println("发送成功!"); session.commit(); session.close(); connection.close(); } /** * 在指定的会话上,通过指定的消息生产者发出一条消息 * * @param session * 消息会话 * @param producer * 消息生产者 */ public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException { // 创建一条文本消息 TextMessage message = session.createTextMessage("Hello ActiveMQ!" + i); // 通过消息生产者发出消息 producer.send(message); } }
消费者
消费完数据之后必须手动提交session
package com.wn.ddd; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class JmsReceiver { public static void main(String[] args) throws JMSException { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); // JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); connection.start(); // Session: 一个发送或接收消息的线程 true:表单开启事务 AUTO_ACKNOWLEDGE:代表自动签收 Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // Destination :消息的目的地;消息发送给谁. // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置 Queue queue = session.createQueue("my-queue"); // 消费者,消息接收者 MessageConsumer consumer = session.createConsumer(queue); while (true) { //receive():获取消息 TextMessage message = (TextMessage) consumer.receive(); if (null != message) { System.out.println("收到消息:" + message.getText()); session.commit(); } else { break; } } //回收资源 session.close(); connection.close(); } }
测试
1.测试在消费数据的时候不commit提交session
1.1 启动生产者
查看队列中的情况
1.2 启动消费者
这里不手动提交session
控制台中可以正确接收到数据,但是队列中的数据就不是正确的
2.正常提交(生产者和消费者都手动提交session)
不带事务session的案例
1.自动签收
2.手动签收
生产者
package com.wn.ddd; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Producter { public static void main(String[] args) throws JMSException { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); // JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); //启动连接 connection.start(); // Session: 一个发送或接收消息的线程 false:代表不带事务的session AUTO_ACKNOWLEDGE:代表自动签收 /* Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);*/ Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE); // Destination :消息的目的地;消息发送给谁. // 获取session注意参数值my-queue是Query的名字 Queue queue = session.createQueue("my-queue"); // MessageProducer:创建消息生产者 MessageProducer producer = session.createProducer(queue); // 设置不持久化 PERSISTENT:代表持久化 NON_PERSISTENT:代表不持久化 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 发送消息 for (int i = 1; i <= 5; i++) { sendMsg(session, producer, i); } System.out.println("发送成功!"); session.close(); connection.close(); } /** * 在指定的会话上,通过指定的消息生产者发出一条消息 * * @param session * 消息会话 * @param producer * 消息生产者 */ public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException { // 创建一条文本消息 TextMessage message = session.createTextMessage("Hello ActiveMQ!" + i); // 通过消息生产者发出消息 producer.send(message);
message.acknowledge(); //手动提交
} }
消费者
package com.wn.ddd; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import sun.plugin2.os.windows.SECURITY_ATTRIBUTES; import javax.jms.*; public class JmsReceiver { public static void main(String[] args) throws JMSException { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); // JMS 客户端到JMS Provider 的连接 Connection connection = connectionFactory.createConnection(); connection.start(); // Session: 一个发送或接收消息的线程 true:表单开启事务 AUTO_ACKNOWLEDGE:代表自动签收 /*Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);*/ Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE); // Destination :消息的目的地;消息发送给谁. // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置 Queue queue = session.createQueue("my-queue"); // 消费者,消息接收者 MessageConsumer consumer = session.createConsumer(queue); while (true) { //receive():获取消息 TextMessage message = (TextMessage) consumer.receive(); if (null != message) { System.out.println("收到消息:" + message.getText()); message.acknowledge(); //手动提交 } else { break; } } //回收资源 session.close(); connection.close(); } }
测试
启动生产者
启动消费者
如果没有手动签收,则会出现和没有commit提交session的情况一样,都是已经消费完的消息,没有情况,造成多次消费。
原文地址:https://www.cnblogs.com/wnwn/p/12307608.html