3-JMS规范和落地产品

package activeMQ;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
 * @Author: chenyawei
 * @Date: 2020-06-02  19:23
 * @Description:
 */
public class JmsProducer_Topic_Persist {
    private static final String ACTIVEMQ_URL = "tcp://192.168.212.140:61616";
    private static final String ACTIVEMQ_TOPIC_NAME = "Topic-Persist";

    public static void main(String[] args) throws JMSException {

        //1.创建连接工厂，按照给定的URL，采用默认的用户名密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2.通过连接工厂,持久化的topic必须在生产者创建并设置持久化完成后调用start
        Connection connection = activeMQConnectionFactory.createConnection();
        //3.创建会话session        //两个参数transacted=事务,acknowledgeMode=确认模式(签收)
        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //4.创建目的地(具体是队列queue还是主题topic)
        Topic topic = session.createTopic(ACTIVEMQ_TOPIC_NAME);
        //5.创建消息的生产者
        MessageProducer messageProducer = session.createProducer(topic);
        //6.设置生产者生产持久化的Topic
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
        //7.启动连接
        connection.start();
        //8.通过使用持久化Topic消息生产者,生产三条消息,发送到MQ的队列里面
        for(int i = 0; i<3; i++){
            System.out.println("****TOPIC_NAME消息发布到MQ完成");
        }

    }
}

package activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.io.IOException;

/**
 * @Author: chenyawei
 * @Date: 2020-06-02  19:40
 * @Description:
 */
public class Jms_Topic_Consumer_Persist {
    private static final String ACTIVEMQ_URL = "tcp://192.168.212.140:61616";
    private static final String ACTIVEMQ_TOPIC_NAME = "Topic-Persist";

    /**
     *  一定要先运行一次消费者,类似于像MQ注册,我订阅了这个主题
     *  然后再运行主题生产者
     *  无论消费着是否在线,都会接收到,在线的立即接收到,不在线的等下次上线把没接收到的接收
     *
     * */
    public static void main(String[] args) throws JMSException, IOException {
        System.out.println("我是3号消费者王五");
        //1.创建连接工厂，按照给定的URL，采用默认的用户名密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2.通过连接工厂,获得connection,设置connectionID
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.setClientID("王五");
        //3.创建会话session
        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //4.创建目的地(具体是队列queue还是主题topic)
        Topic topic = session.createTopic(ACTIVEMQ_TOPIC_NAME);
        //5.通过session创建持久化订阅
        TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic,"我是王五");
        //6.启动连接
        connection.start();
        //7.接收消息
        topicSubscriber.setMessageListener(message -> {
            if (message instanceof TextMessage){
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("收到的持久化订阅消息: " + textMessage.getText());
                } catch (JMSException e)
                {
                    e.printStackTrace();
                }
            }
        });

    }
}

package activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @Author: chenyawei
 * @Date: 2020-06-08  15:40
 * @Description:
 */
public class Jms_TX_Producer {

    private static final String ACTIVEMQ_URL = "tcp://192.168.212.141:61616";
    private static final String ACTIVEMQ_QUEUE_NAME = "Queue-TX";
    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂，按照给定的URL，采用默认的用户名密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2.通过连接工厂,获得connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3.创建会话session
        //两个参数  truetransacted=事务,acknowledgeMode=确认模式(签收)
        //false: 只要执行send，就进入到队列中；关闭事务，那第2个签收参数的设置需要有效。
        //true：先执行send再执行commit，消息才被真正提交到队列中；消息需要需要批量提交，需要缓冲处理。
        //开启事务需要commit
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        //4.创建目的地(具体是队列queue还是主题topic)
        Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);
        //5.创建消息的生产者,并设置不持久化消息
        MessageProducer producer = session.createProducer(queue);
        //6.通过使用消息生产者,生产三条消息,发送到MQ的队列里面
        try {
            for (int i = 0; i<3; i++) {
                TextMessage textMessage = session.createTextMessage("tx msg--" + i);
                producer.send(textMessage);
            }
            //7.提交事务
            session.commit();
            System.out.println("消息发送完成");
        } catch (Exception e) {
            System.out.println("出现异常,消息回滚");
            session.rollback();
        } finally {
            //8.关闭资源
            producer.close();
            session.close();
            connection.close();
        }
}
}

package activeMQ;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

/**
 * @Author: chenyawei
 * @Date: 2020-06-08  15:48
 * @Description: 
 */
public class Jms_TX_Consumer {

    private static final String ACTIVEMQ_URL = "tcp://192.168.212.141:61616";
    private static final String ACTIVEMQ_QUEUE_NAME = "Queue-TX";

    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂，按照给定的URL，采用默认的用户名密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2.通过连接工厂,获得connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3.创建会话session
        // 两个参数transacted=事务,acknowledgeMode=确认模式(签收)
        // 消费者开启了事务就必须手动提交，不然会重复消费消息
        Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
        //4.创建目的地(具体是队列queue还是主题topic)
        Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);
        //5.创建消息的消费者,指定消费哪一个队列里面的消息
        MessageConsumer messageConsumer = session.createConsumer(queue);
        //6.通过监听的方式消费消息
        messageConsumer.setMessageListener(new MessageListener() {
            int a = 0;
            @Override
            public void onMessage(Message message) {
                if (message instanceof TextMessage){
                    try {
                        if (a == 2){
                            System.out.println(1/0);
                        }
                        TextMessage textMessage = (TextMessage) message;
                        System.out.println("***消费者接收到的消息:   " + textMessage.getText());
                        session.commit();
                        a = a + 1;
                    }catch (Exception e){
                        System.out.println("出现异常，消费失败，放弃消费");
                    }

                    try {
                        session.rollback();
                        a=0;
                    } catch (JMSException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        });
    }
}