public class JmsProduceJDBC {
    public static final String ACTIVEMQ_URL = "tcp://localhost:61616";

    public static final String USERNAME = "admin";

    public static final String PASSWORD = "hll123";

    public static final String QUEUE_NAME = "jdbc01";

    public static void main(String[] args) throws Exception {
        //1.按照给定的url创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD,ACTIVEMQ_URL);
        // 2.通过工厂连接connection 和启动
        Connection connection = activeMQConnectionFactory.createConnection();
        // 3.启动
        connection.start();
        // 4.创建会话session
        //两个参数，第一个事务，第二个签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5.创建目的地，队列、主题，这里用队列
        Queue queue = session.createQueue(QUEUE_NAME);
        // 6.创建消息的生产者
        MessageProducer messageProducer = session.createProducer(queue);

        /**
         * 持久化必须设置,一定要开启！！！
         */
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

        // 7.通过MessageProducer生产3条消息发送到消息队列中
        for (int i = 1; i <= 6; i++) {
            //8.创建消息
            TextMessage textMessage = session.createTextMessage("msg:" + LocalDateTime.now());
            //9.发送消息
            messageProducer.send(textMessage);
        }

        // 10.关闭资源
        messageProducer.close();
        session.close();
        connection.close();

        System.out.println(" **** 消息发送到MQ完成 **** ");
    }
}

public class JmsConsumerJDBC {
    public static final String ACTIVEMQ_URL = "tcp://localhost:61616";

    public static final String USERNAME = "admin";

    public static final String PASSWORD = "hll123";

    public static final String QUEUE_NAME = "jdbc01";

    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, ACTIVEMQ_URL);
        // 创建连接connection
        Connection connection = activeMQConnectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建会话session
        Session session = connection.createSession(false,  Session.AUTO_ACKNOWLEDGE);
        //创建队列，同生产者一致
        Queue queue = session.createQueue(QUEUE_NAME);
        //创建消息消费者
        MessageConsumer messageConsumer = session.createConsumer(queue);

        /**
         * 方法2：通过监听器的方式
         */
        messageConsumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if (null != message && message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("**** 消费者接收到消息 ****：" + textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        System.in.read(); // 必须加行代码，不然程序会直接往下执行结束了
        messageConsumer.close();
        session.close();
        connection.close();
        System.out.println("**** 消费者消费消息完成 ****");


    }
}

public class JmsConsumerTopicJDBC {

    public static final String ACTIVEMQ_URL = "tcp://localhost:61616";

    public static final String USERNAME = "admin";

    public static final String PASSWORD = "hll123";

    public static final String TOPIC_NAME = "topic-jdbc";

    public static void main(String[] args) throws Exception {
        /**
         * 持久化主题消息订阅，类似于微信公众号订阅
         * 需要先启动消费者，订阅上主题之后，后续生产主题消息，消费者(订阅者)就会接收到消息
         * 消费者(订阅者)订阅主题之后，不管是在线还是离线状态，只要保持正常订阅状态，期间生产的消息都会接收到。离线的会在再次在线后接收到之前的消息
         */
        System.out.println("jdbc-1"); //模拟订阅用户
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, ACTIVEMQ_URL);
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.setClientID("jdbc-1"); // 设置clientId,表明订阅者

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic(TOPIC_NAME);
        TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "jdbc-1");
        connection.start();

        Message message = topicSubscriber.receive();
        while (null != message) {
            TextMessage textMessage = (TextMessage) message;
            System.out.println("收到的持久化topic消息：" + textMessage.getText());
            message = topicSubscriber.receive();
         }

        session.close();
        connection.close();
    }
}

public class JmsProduceTopicJDBC {
    public static final String ACTIVEMQ_URL = "tcp://localhost61616";

    public static final String USERNAME = "admin";

    public static final String PASSWORD = "hll123";

    public static final String TOPIC_NAME = "topic-jdbc";


    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, ACTIVEMQ_URL);
        Connection connection = activeMQConnectionFactory.createConnection();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic(TOPIC_NAME);
        MessageProducer messageProducer = session.createProducer(topic);
        // connection启动之前必须设置持久化主题
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

        connection.start();

        for (int i = 1; i <= 3; i++) {
            TextMessage textMessage = session.createTextMessage("jdbc-msg:" + i);
            messageProducer.send(textMessage);
        }

        messageProducer.close();
        session.close();
        connection.close();
        System.out.println(" **** 持久化消息发送到MQ完成 **** ");
    }
}