用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(); |
|