用ActiveMQ遇到的消息确认问题

问题：我的ActiveMQ接收消息用的是topic模式，持久化订阅，问题是我用了JMS接收消息的代码每次重新启动总是会收到最后一次的消息，但这些消息是已经接收过了的，而且启动一次就收到一次，难道ActiveMQ不会清除缓存的吗？

//创建JMS连接和会话
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);   
connection = factory.createConnection();   
connection.setClientID(Constant.JMS_CLIENT_ID);   
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
// 创建消息发送主题和发送者
Topic jmsSendTopic = session.createTopic(sendTopic);
sendTopicProducer = session.createProducer(jmsSendTopic);
sendTopicProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
sendTopicProducer.setTimeToLive(Message.DEFAULT_TIME_TO_LIVE);
// 创建消息接收主题和接收者
Topic jmsReceiveTopic = session.createTopic(receiveTopic);
receiveTopicConsumer = session.createDurableSubscriber(jmsReceiveTopic,Constant.JMS_SUBSCRIBE_NAME);
receiveTopicConsumer.setMessageListener(this);
connection.start();  

解答：问题原因在于这段代码在接收到JMS消息时不会向ActiveMQ服务器确认消息的接收，故而ActiveMQ服务器一直认为该消息没有成功发送给接收者，因而每次接收者重启之后就会收到ActiveMQ服务器发送过来的消息。在这里要解释一下session的创建。

session = connection.createSession(true,Session.Auto_ACKNOWLEDGE);

当createSession第一个参数为true时，表示创建的session被标记为transactional的，确认消息就通过确认和校正来自动地处理，第二个参数应该是没用的。

session = connection.createSession(false,Session.Auto_ACKNOWLEDGE);

当createSession的第一个参数为false时，表示创建的session没有标记为transactional，此时有三种用于消息确认的选项：
**AUTO_ACKNOWLEDGE session将自动地确认收到的一则消息;
**CLIENT_ACKNOWLEDGE 客户端程序将确认收到的一则消息，调用这则消息的确认方法;
**DUPS_OK_ACKNOWLEDGE 这个选项命令session“懒散的”确认消息传递，可以想到，这将导致消息提供者传递的一些复制消息可能出错。

JMS有两种消息传递方式。标记为NON_PERSISTENT的消息最多传递一次，而标记为PERSISTENT的消息将使用暂存后再转发的机理投递。如果一个JMS服务离线，那么持久性消息不会丢失，但是得等到这个服务恢复联机的时候才会被传递。所以默认的消息传递方式是非持久性的，虽然使用非持久性消息可能降低内存和需要的存储器，但这种传递方式只有当你不需要接收所有消息时才使用。
因此正确的代码只需改动一处就行了，即将true改为false

//创建JMS连接和会话
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);   
connection = factory.createConnection();   
connection.setClientID(Constant.JMS_CLIENT_ID);   
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建消息发送主题和发送者
Topic jmsSendTopic = session.createTopic(sendTopic);
sendTopicProducer = session.createProducer(jmsSendTopic);
sendTopicProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
sendTopicProducer.setTimeToLive(Message.DEFAULT_TIME_TO_LIVE);
// 创建消息接收主题和接收者
Topic jmsReceiveTopic = session.createTopic(receiveTopic);
receiveTopicConsumer = session.createDurableSubscriber(jmsReceiveTopic,Constant.JMS_SUBSCRIBE_NAME);
receiveTopicConsumer.setMessageListener(this);
connection.start();  

评论

1 楼 liumuqi110 2012-12-21  

你的结论是错误的吧。。这是因为事务问题。你没提交  这个就会滚 从发你了