ActiveMQ消息签收机制

Posted 2021-03-09 lowerma

 消费者客户端成功接收一条消息的标志是：这条消息被签收。
         消费者客户端成功接收一条消息一般包括三个阶段：
         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.TRUE, 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.TRUE, Session.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();
    }
}