时间:2021-05-21
这篇文章主要介绍了ActiveMQ消息签收机制代码实例解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
消费者客户端成功接收一条消息的标志是:这条消息被签收。
消费者客户端成功接收一条消息一般包括三个阶段:
1、消费者接收消息,也即从MessageConsumer的receive方法返回
2、消费者处理消息
3、消息被签收
其中,第三阶段的签收可以有ActiveMQ发起,也可以由消费者客户端发起,取决于Session是否开启事务以及签收模式的设置。
在带事务的Session中,消费者客户端事务提交之时,消息自动完成签收。
在不带事务的Session中,消息何时以及如何被签收取决于Session的签收模式设置
非事务Session可以设置如下几种签收模式:
1.Session.AUTO_ACKNOWLEDGE
当消息从MessageConsumer的receive方法返回或者从MessageListener接口的onMessage方法返回时,会话自动确认消息签收
2.Session.CLIENT_ACKNOWLEDGE
需要消费者客户端主动调用acknowledge方法签收消息,这种模式实在Session层面进行签收的,签收一个已经消费的消息会自动的签收这个Session已消费的所有消息:
例如一个消费者在一个Session中消费了5条消息,然后确认第3条消息,所有这5条消息都会被签收
3.Session.DUPS_OK_ACKNOWLEDGE
这种方式允许JMS不必急于确认收到的消息,允许在收到多个消息之后一次完成确认,与Auto_AcKnowledge相比,这种确认方式在某些情况下可能更有效,因为没有确认,当系统崩溃或者网络出现故障的时候,消息可以被重新传递.
这种方式会引起消息的重复,但是降低了Session的开销,所以只有客户端能容忍重复的消息才可使用。(如果ActiveMQ再次传送同一消息,那么消息头中的JMSRedelivered将被设置为true)
带事务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(); }}不带事务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.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.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(); }}以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
这篇文章主要介绍了ActiveMQ持久化机制代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下用户注册成
实现消息队列的两种方式ApacheActiveMQ官方实例发送消息直接在Apache官网http://activemq.apache.org/download-
本文实例讲述了Android编程中的消息机制。分享给大家供大家参考,具体如下:在分析Android消息机制之前,我们先来看一段代码:publicclassMai
本文实例讲述了ActiveMQ在C#中的应用。分享给大家供大家参考,具体如下:ActiveMQ是个好东东,不必多说。ActiveMQ提供多种语言支持,如Java
Java反射机制的实例详解前言今天介绍下Java的反射机制,以前我们获取一个类的实例都是使用new一个实例出来。那样太low了,今天跟我一起来学习学习一种更加高