JMS消息的可靠性机制

Posted wnwn

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了JMS消息的可靠性机制相关的知识,希望对你有一定的参考价值。

ActiveMQ消息签收机制:

  客户端成功接收一条消息的标志是一条消息被签收,成功应答。

消息的签收请求分为两种:

  1.带事务的session

    如果session带有事务,并且事务成功提交,则消息被自动签收。如果事务回滚,则消息会被再次传送。

  2.不带事务的session

    不带事务的session的签收方式,取决于session的配置

    ActiveMQ支持以下三种模式:

      Seesion.AUTO_ACKNOWLEDGE:消息自动签收;

      Session.CLIENT_ACKNOWLEDGE:客户端调用acknowledge方法手动签收

        textMessage.acknowledge();//手动签收

      Session.DUPS_OK_ACKNOWLEDGE:不是必须签收,消息可能会重复发送。在第二次重新传送消息的时候,消息只有在被确认之后,才认为已经被成功地消费了。消息的成功消费通常包含三个阶段:客户端接收消息客户处理消息和消息被确认。在事务性会话中,当一个事务提交的时候,确认自动发生。在非事务性会话中,消息何时被确认取决于创建会话时的应答模式。

带事务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();
    }
}

  测试

    1.测试在消费数据的时候不commit提交session

      1.1 启动生产者

        技术图片

        查看队列中的情况

        技术图片

      1.2 启动消费者

        这里不手动提交session

        技术图片

        控制台中可以正确接收到数据,但是队列中的数据就不是正确的

        技术图片

    2.正常提交(生产者和消费者都手动提交session)

        技术图片

不带事务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();
    }
}

    测试

      启动生产者

        技术图片

      启动消费者

        技术图片

         如果没有手动签收,则会出现和没有commit提交session的情况一样,都是已经消费完的消息,没有情况,造成多次消费。

 

以上是关于JMS消息的可靠性机制的主要内容,如果未能解决你的问题,请参考以下文章

activeMq-JMS消息可靠性机制-4

ActiveMQ(03):JMS的可靠性机制

JMS与消息队列

Jms规范学习

activeMQ从入门到简单集群指南

JMS