这篇文章主要介绍了ActiveMQ消息签收机制代码实例解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
消费者客户端成功接收一条消息的标志是:这条消息被签收。
消费者客户端成功接收一条消息一般包括三个阶段:
1,消费者接收消息,也即从MessageConsumer的接收方法返回
2,消费者处理消息
3,消息被签收
其中,第三阶段的签收可以有ActiveMQ发起,也可以由消费者客户端发起,取决于会话是否开启事务以及签收模式的设置。
在带事务的会话中,消费者客户端事务提交之时,消息自动完成签收。
在不带事务的会话中,消息何时以及如何被签收取决于会话的签收模式设置
非事务会议可以设置如下几种签收模式:
1.。AUTO_ACKNOWLEDGE
当消息从MessageConsumer的接收方法返回或者从MessageListener接口的onMessage方法返回时,会话自动确认消息签收
2.。CLIENT_ACKNOWLEDGE
需要消费者客户端主动调用承认方法签收消息,这种模式实在会话层面进行签收的,签收一个已经消费的消息会自动的签收这个会议已消费的所有消息:
例如一个消费者在一个会话中消费了5条消息,然后确认第3条消息,所有这5条消息都会被签收
3.。DUPS_OK_ACKNOWLEDGE
这种方式允许JMS不必急于确认收到的消息,允许在收到多个消息之后一次完成确认,与Auto_AcKnowledge相比,这种确认方式在某些情况下可能更有效,因为没有确认,当系统崩溃或者网络出现故障的时候,消息可以被重新传递。
这种方式会引起消息的重复,但是降低了会话的开销,所以只有客户端能容忍重复的消息才可使用。(如果ActiveMQ再次传送同一消息,那么消息头中的JMSRedelivered将被设置为true)
生产者
必须在生产完数据之后手动提交会议
包com.wn.ddd;
进口org.apache.activemq.ActiveMQConnection;
进口org.apache.activemq.ActiveMQConnectionFactory;
进口javax.jms。*;
公开课生产者{
公共静态void main (String [] args)抛出JMSException {//ConnectionFactory:连接工厂、JMS用它创建连接
ConnectionFactory ConnectionFactory=new ActiveMQConnectionFactory (ActiveMQConnection.DEFAULT_USER
ActiveMQConnection。DEFAULT_PASSWORD,“tcp://127.0.0.1:61616”);//JMS客户端到JMS提供者的连接
连接连接=connectionFactory.createConnection ();//启动连接
connection.start ();//会话:一个发送或接收消息的线程错误:代表不带事务的会话AUTO_ACKNOWLEDGE:代表自动签收
会议会话=connection.createSession(布尔。真的,Session.AUTO_ACKNOWLEDGE);//目的地:消息的目的地;消息发送给谁。//获取会话注意参数值我的队列是查询的名字
队列的队列=session.createQueue(“我的队列”);//MessageProducer:创建消息生产者
MessageProducer生产商=session.createProducer(队列);//设置不持久化持久:代表持久化NON_PERSISTENT:代表不持久化
producer.setDeliveryMode (DeliveryMode.PERSISTENT);//发送消息
for (int i=1;我& lt;=5;我+ +){
sendMsg(会话,制片人,我);
}
System.out.println(“发送成功!”);
session.commit ();
session.close ();
connection.close ();
}/* *
*在指定的会话上,通过指定的消息生产者发出一条消息
*
* @param会话
*消息会话
* @param生产商
*消息生产者
*/公共静态孔隙sendMsg(会议会话,MessageProducer生产国,int i)抛出JMSException {//创建一条文本消息
TextMessage消息=会话。createTextMessage(“你好ActiveMQ !”+ i);//通过消息生产者发出消息
producer.send(消息);
}
}
消费者
消费完数据之后必须手动提交会议
包com.wn.ddd;
进口org.apache.activemq.ActiveMQConnection;
进口org.apache.activemq.ActiveMQConnectionFactory;
进口javax.jms。*;
公开课JmsReceiver {
公共静态void main (String [] args)抛出JMSException {//ConnectionFactory:连接工厂、JMS用它创建连接
ConnectionFactory ConnectionFactory=new ActiveMQConnectionFactory (ActiveMQConnection.DEFAULT_USER
ActiveMQConnection。DEFAULT_PASSWORD,“tcp://127.0.0.1:61616”);//JMS客户端到JMS提供者的连接
连接连接=connectionFactory.createConnection ();
connection.start ();//会话:一个发送或接收消息的线程真的:表单开启事务AUTO_ACKNOWLEDGE:代表自动签收
会议会话=connection.createSession(布尔。真的,Session.AUTO_ACKNOWLEDGE);//目的地:消息的目的地;消息发送给谁。//获取会话注意参数值xingbo.xu-queue是一个服务器的队列,须在在ActiveMq的控制台配置
队列的队列=session.createQueue(“我的队列”);//消费者,消息接收者
MessageConsumer消费=session.createConsumer(队列);
而(真){//接收():获取消息
TextMessage消息=(TextMessage) consumer.receive ();
如果(零!=消息){
system . out。println(“收到消息:“+ message.getText ());
session.commit ();
其他}{
打破;
}
}//回收资源
session.close ();
connection.close ();
}
}
