3.1 JMS消息确认机制

在 Session 接口中定义的几个常量：

• AUTO_ACKNOWLEDGE = 1 自动确认

• CLIENT_ACKNOWLEDGE = 2 客户端手动确认

• DUPS_OK_ACKNOWLEDGE = 3 自动批量确认

• SESSION_TRANSACTED = 0 事务提交并确认

3.2 代码实现

消息消费端在创建 Session 对象时需要指定应答模式为客户端手动应答，当消费者获取到消息并成功处理后需要调用 message.acknowledge() 方法进行应答，通知 Broker 消费成功。

如果处理过程中出现异常，需要使用 session.recover() 通知 Broker 重复消息，默认最多重复  6 次。

• 首先需要启动 ActiveMQ

• 打开 IDEA，新建一个 Maven 工程

• 在 pom.xml 文件中导入依赖的 jar

<dependency>            <groupId>org.apache.activemqgroupId>            <artifactId>activemq-clientartifactId>            <version>5.15.13version>        dependency>        <dependency>            <groupId>junitgroupId>            <artifactId>junitartifactId>            <version>4.13version>            <scope>testscope>        dependency>

• 创建消费发送者和消息接收者

  • 消息发送者

//消息发送者    @Test    public void testProvider() throws Exception {        //创建连接工厂对象        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");        //在工厂中获取一个连接对象        Connection connection = connectionFactory.createConnection();        //连接 MQ 服务        connection.start();        //获取 Session 对象 Session.AUTO_ACKNOWLEDGE 设置为自动应答        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);        //通过 Session 对象创建一个主题        Topic topic = session.createTopic("sea");        //通过 session 对象创建消息发送者        MessageProducer producer = session.createProducer(topic);        //通过  session 创建消息对象        Message message = session.createTextMessage("我的家在东北");        //发送消息        producer.send(message);        //关闭资源        producer.close();        session.close();        connection.close();    }

• 消息接收者

//主消息接收方    @Test    public void testReceiver() throws Exception {        //创建连接工厂对象        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");        //在工厂中获取一个连接对象        Connection connection = connectionFactory.createConnection();        //连接 MQ 服务        connection.start();        //获取 Session 对象 Session.AUTO_ACKNOWLEDGE 设置为自动应答        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);        //通过 Session 对象创建一个主题        Topic topic = session.createTopic("sea");        //通过 session 对象创建消息消费者        MessageConsumer consumer = session.createConsumer(topic);        //创建消息监听器        consumer.setMessageListener(new MessageListener() {            //当监听的主题中存在消息，此时这个方法自动执行            @Override            public void onMessage(Message message) {                TextMessage textMessage = (TextMessage) message;                try {                    System.out.println("此时监听到的消息为:" + textMessage.getText());                } catch (JMSException e) {                    e.printStackTrace();                }            }        });        System.in.read();    }

我们先启动消费接收者，叫其一直处于监听状态，然后运行消息发送者：

可见此时消息发送者成功将消息发出且被接收者成功订阅，然而此时我们并没有考虑到消息被监听失败的场景，如果消息发送失败，那么我们需要重发。此时将消费端的代码改动如下：

可以进行失败次数重发的代码：

//主消息接收方    @Test    public void testReceiver() throws Exception {        //创建连接工厂对象        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");        //在工厂中获取一个连接对象        Connection connection = connectionFactory.createConnection();        //连接 MQ 服务        connection.start();        //获取 Session 对象 Session.CLIENT_ACKNOWLEDGE 设置为手动应答，避确保失败后可以进行重发        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);        //通过 Session 对象创建一个主题        Topic topic = session.createTopic("sea");        //通过 session 对象创建消息消费者        MessageConsumer consumer = session.createConsumer(topic);        //创建消息监听器        consumer.setMessageListener(new MessageListener() {            //当监听的主题中存在消息，此时这个方法自动执行            @Override            public void onMessage(Message message) {                TextMessage textMessage = (TextMessage) message;                try {                    if (textMessage.getText().equals("666")) {                        //客户端手动应答                        message.acknowledge();                    } else {                        System.out.println("消息监听失败，请重新发送~");                        session.recover();                        int i = 1 / 0;                    }                } catch (JMSException e) {                    e.printStackTrace();                }            }        });        System.in.read();    }

此时的运行结果如下：