ActiveMQ消息签收机制代码实例详解
这篇文章主要介绍了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
packagecom.wn.ddd;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
importjavax.jms.*;
publicclassProducter{
publicstaticvoidmain(String[]args)throwsJMSException{
//ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,"tcp://127.0.0.1:61616");
//JMS客户端到JMSProvider的连接
Connectionconnection=connectionFactory.createConnection();
//启动连接
connection.start();
//Session：一个发送或接收消息的线程false:代表不带事务的sessionAUTO_ACKNOWLEDGE：代表自动签收
Sessionsession=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//Destination：消息的目的地;消息发送给谁.
//获取session注意参数值my-queue是Query的名字
Queuequeue=session.createQueue("my-queue");
//MessageProducer：创建消息生产者
MessageProducerproducer=session.createProducer(queue);
//设置不持久化PERSISTENT:代表持久化NON_PERSISTENT：代表不持久化
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
//发送消息
for(inti=1;i<=5;i++){
sendMsg(session,producer,i);
}
System.out.println("发送成功！");
session.commit();
session.close();
connection.close();
}
/**
*在指定的会话上，通过指定的消息生产者发出一条消息
*
*@paramsession
*消息会话
*@paramproducer
*消息生产者
*/
publicstaticvoidsendMsg(Sessionsession,MessageProducerproducer,inti)throwsJMSException{
//创建一条文本消息
TextMessagemessage=session.createTextMessage("HelloActiveMQ！"+i);
//通过消息生产者发出消息
producer.send(message);
}
}
消费者
消费完数据之后必须手动提交session
packagecom.wn.ddd;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
importjavax.jms.*;
publicclassJmsReceiver{
publicstaticvoidmain(String[]args)throwsJMSException{
//ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,"tcp://127.0.0.1:61616");
//JMS客户端到JMSProvider的连接
Connectionconnection=connectionFactory.createConnection();
connection.start();
//Session：一个发送或接收消息的线程true：表单开启事务AUTO_ACKNOWLEDGE：代表自动签收
Sessionsession=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//Destination：消息的目的地;消息发送给谁.
//获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
Queuequeue=session.createQueue("my-queue");
//消费者，消息接收者
MessageConsumerconsumer=session.createConsumer(queue);
while(true){
//receive():获取消息
TextMessagemessage=(TextMessage)consumer.receive();
if(null!=message){
System.out.println("收到消息："+message.getText());
session.commit();
}else{
break;
}
}
//回收资源
session.close();
connection.close();
}
}
不带事务session的案例
1.自动签收
2.手动签收
生产者
packagecom.wn.ddd;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
importjavax.jms.*;
publicclassProducter{
publicstaticvoidmain(String[]args)throwsJMSException{
//ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,"tcp://127.0.0.1:61616");
//JMS客户端到JMSProvider的连接
Connectionconnection=connectionFactory.createConnection();
//启动连接
connection.start();
//Session：一个发送或接收消息的线程false:代表不带事务的sessionAUTO_ACKNOWLEDGE：代表自动签收
/*Sessionsession=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);*/
Sessionsession=connection.createSession(Boolean.FALSE,Session.CLIENT_ACKNOWLEDGE);
//Destination：消息的目的地;消息发送给谁.
//获取session注意参数值my-queue是Query的名字
Queuequeue=session.createQueue("my-queue");
//MessageProducer：创建消息生产者
MessageProducerproducer=session.createProducer(queue);
//设置不持久化PERSISTENT:代表持久化NON_PERSISTENT：代表不持久化
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
//发送消息
for(inti=1;i<=5;i++){
sendMsg(session,producer,i);
}
System.out.println("发送成功！");
session.close();
connection.close();
}
/**
*在指定的会话上，通过指定的消息生产者发出一条消息
*
*@paramsession
*消息会话
*@paramproducer
*消息生产者
*/
publicstaticvoidsendMsg(Sessionsession,MessageProducerproducer,inti)throwsJMSException{
//创建一条文本消息
TextMessagemessage=session.createTextMessage("HelloActiveMQ！"+i);
//通过消息生产者发出消息
producer.send(message);
message.acknowledge();//手动提交
}
}
消费者
packagecom.wn.ddd;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
importsun.plugin2.os.windows.SECURITY_ATTRIBUTES;
importjavax.jms.*;
publicclassJmsReceiver{
publicstaticvoidmain(String[]args)throwsJMSException{
//ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,"tcp://127.0.0.1:61616");
//JMS客户端到JMSProvider的连接
Connectionconnection=connectionFactory.createConnection();
connection.start();
//Session：一个发送或接收消息的线程true：表单开启事务AUTO_ACKNOWLEDGE：代表自动签收
/*Sessionsession=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);*/
Sessionsession=connection.createSession(Boolean.FALSE,Session.CLIENT_ACKNOWLEDGE);
//Destination：消息的目的地;消息发送给谁.
//获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
Queuequeue=session.createQueue("my-queue");
//消费者，消息接收者
MessageConsumerconsumer=session.createConsumer(queue);
while(true){
//receive():获取消息
TextMessagemessage=(TextMessage)consumer.receive();
if(null!=message){
System.out.println("收到消息："+message.getText());
message.acknowledge();//手动提交
}else{
break;
}
}
//回收资源
session.close();
connection.close();
}
}
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。