ActiveMQ学习之消息事务与签收

1、事务transaction

           一、事务偏生产者/签收偏消费者java

            二、生产者提交事务只有两个状态true/falsespring

                    false:只须要执行send方法,消息就会进入队列中apache

                                关闭事务,那第二个参数签收的设置须要有效springboot

                     true:先执行send方法,在执行commit,消息才被真正的提交到队列中session

                                消息须要批量发送,须要缓冲区处理tcp

            未开启事生产者url

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/2 17:04
 * @Version: 1.0
 */
public class ActiveMQTest {
    //url路径
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //队列名称
    private static final String QUEUE_NAME="queue01";
   

    public static void main(String[] args) {
        //一、建立链接工厂
        //若是帐号密码没有修改的话,帐号密码默认均为admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帐号密码修改的话
        //第一个参数为帐号,第二个为密码,第三个为请求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、经过链接工厂获取链接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session会话
            //里面会有两个参数,第一个为事务,第二个是签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、建立目的地(具体是队列仍是主题),这里是建立队列
            Queue queue=session.createQueue(QUEUE_NAME);
            //五、建立消息生产者,队列模式
            MessageProducer messageProducer = session.createProducer(queue);
            //六、经过messageProducer生产三条消息发送到MQ消息队列中
            for (int i=0;i<3;i++){
                //七、建立消息
                TextMessage textMessage = session.createTextMessage("msg----->" + i);//建立一个文本消息
                //消息属性
                textMessage.setStringProperty("c01","vip");
                //八、经过messageProducer发送给mq
                messageProducer.send(textMessage);
                //九、数据非持久化
                messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
            }
            messageProducer.close();
            session.close();
            connection.close();
            System.out.println("消息发送成功");
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

                已开启事生产者spa

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/2 17:04
 * @Version: 1.0
 */
public class ActiveMQTest {
    //url路径
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //队列名称
    private static final String QUEUE_NAME="queue01";


    public static void main(String[] args) {
        //一、建立链接工厂
        //若是帐号密码没有修改的话,帐号密码默认均为admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帐号密码修改的话
        //第一个参数为帐号,第二个为密码,第三个为请求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、经过链接工厂获取链接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session会话
            //里面会有两个参数,第一个为事务,第二个是签收
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //四、建立目的地(具体是队列仍是主题),这里是建立队列
            Queue queue=session.createQueue(QUEUE_NAME);
            //五、建立消息生产者,队列模式
            MessageProducer messageProducer = session.createProducer(queue);
            //六、经过messageProducer生产三条消息发送到MQ消息队列中
            for (int i=0;i<3;i++){
                //七、建立消息
                TextMessage textMessage = session.createTextMessage("msg----->" + i);//建立一个文本消息
                //消息属性
                textMessage.setStringProperty("c01","vip");
                //八、经过messageProducer发送给mq
                messageProducer.send(textMessage);
                //九、数据非持久化
                messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
            }
            messageProducer.close();
            session.commit();
            session.close();
            connection.close();
            System.out.println("消息发送成功");
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

2、签收

           1、非事务签收code

                   一、自动签收(默认)队列

                    二、手动签收

                                Session.CLIENT_ACKNOWLEDGE,须要客户端调用acknowledge(message.acknowledge();)方法进行手动签收

                    三、容许重复消息

                                Session.DUPS_OK_ACKNOWLEDGE

                    四、事务级

                                Session.SESSION_TRANSACTED

           2、签收

                   一、非事务手动签收

                            须要客户端调用acknowledge(message.acknowledge();)方法进行手动签收,若是不签收,消息重复消费

                    二、开启事务,设置手动签收

                            有commit语句的状况下,不须要调用message.acknowledge();方法进行签收,开启事务会认为你自动签收

                            没有commit语句的状况下,即便调用message.acknowledge();方法进行签收,也会出现重复消费

3、总结

                    一、事务大于签收,因此必定要在开启事务的状况下进行签收,在事务性会话当中,当一个事务被成功提交则消息被自动签收。若是事务回滚,则消息会被再次传送。

                    二、非事务性会话当中,消息什么时候被签收取决于建立会话时的应答模式(message.acknowledge();)

4、MQ消息可靠性:

                一、事物

                二、签收

                三、消息持久性

                 四、集群                   

相关文章
相关标签/搜索