飞天班第41节：ActiveMQ消息队列实战-1

/* @param destination 目的地：queue 或者 topic
     * @param message 
     * @param deliveryMode 是否持久化模式
     * @param priority the priority for this message 优先级 0-4普通消息，5-9紧急消息，默认4级
     * @param timeToLive the message's lifetime (in milliseconds) 生存时间，过期时间
     */
    @Override
    public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
        this.send(destination, message, deliveryMode, priority, timeToLive, null);
    }

package com.coding.queue;

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

public class JmsProduce {
    // 1、消息发送到哪里
    // 39.105.61.80:8161  控制板
    // 39.105.61.80:61616 程序通信tcp协议
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);

        // 3、获得连接
        Connection connection = factory.createConnection();
        connection.start();

        // 4、创建会话
      	// 第一个参数：是否开启事务，不开启，默认自动提交，开启需要手动提交事务
        // 参数：事务、签收  AUTO_ACKNOWLEDGE ： 自动确认
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5、模式，目的地（队列，主题）
        // session.createTopic();
        Queue queue = session.createQueue("queue01");

        // 6、发送消息（生成者）
        MessageProducer producer = session.createProducer(queue);

        // 7、发送具体的消息到队列, 创建的消息类型是什么就些什么
        // 简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。
        for (int i = 0; i < 3; i++) {
            TextMessage message = session.createTextMessage("msg=>" + i);
            // 8、发送消息
            producer.send(message);
        }

        // 9、用完记得关闭资源
        producer.close();
        session.close();
        connection.close();
        System.out.println("发送完毕");
    }
}

package com.coding.queue;

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

public class JmsConsumer {
    // 1、消息发送到哪里
    // 39.105.61.80:8161  控制板
    // 39.105.61.80:61616 程序通信tcp协议
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);

        // 3、获得连接
        Connection connection = factory.createConnection();
        connection.start();

        // 4、创建会话
        // 参数：事务、签收  AUTO_ACKNOWLEDGE ： 自动确认
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5、模式，目的地（队列，主题）
        // session.createTopic();
        Queue queue = session.createQueue("queue01");

        // 6、消息接受者（角色变化）
        MessageConsumer consumer = session.createConsumer(queue);

        // 7、获取消息
        while (true){
            // tcp  等待接收消息的过程。 receive
            TextMessage receive = (TextMessage) consumer.receive();
          // receive 阻塞等待
					// receive(long timeout) 超过多少ms，没有消息可以消费，就消费结束，然后不等待了
					// TextMessage receive = (TextMessage) consumer.receive(1000);
            if (receive!=null){
                System.out.println("接收到消息："+ receive.getText());
            }else {
                break;
            }
        }

        // 9、用完记得关闭资源
        consumer.close();
        session.close();
        connection.close();
        System.out.println("接收完毕");
    }
}

package com.coding.topic;

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

public class JmsProduceTopic {
    // 1、消息发送到哪里
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);

        // 3、获得连接
        Connection connection = factory.createConnection();
        connection.start();

        // 4、创建会话
        // 参数：事务、签收  AUTO_ACKNOWLEDGE ： 自动确认
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5、创建主题 Topic
        Topic topic = session.createTopic("topic01");

        // 6、发送消息（生成者）
        MessageProducer producer = session.createProducer(topic);

        // 7、发送具体的消息到队列, 创建的消息类型是什么就些什么
        // 简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。
        for (int i = 0; i < 6; i++) {
            TextMessage message = session.createTextMessage("msg=>" + i);
            // 8、发送消息
            producer.send(message);
        }

        // 9、用完记得关闭资源
        producer.close();
        session.close();
        connection.close();
        System.out.println("发送完毕");
    }
}

package com.coding.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

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

public class JmsConsumerTopic {
    // 1、消息发送到哪里
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException, IOException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);

        // 3、获得连接
        Connection connection = factory.createConnection();
        connection.start();

        // 4、创建会话
        // 参数：事务、签收  AUTO_ACKNOWLEDGE ： 自动确认
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5、创建主题 Topic
        Topic topic = session.createTopic("topic01");

        // 6、消息接受者（角色变化）
        MessageConsumer consumer = session.createConsumer(topic);

        // 7、获取消息，监听
        consumer.setMessageListener((message) -> {
            if (message!=null && message instanceof TextMessage){
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("接收到主题消息"+textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        // 卡住等待
        System.in.read();

        // 9、用完记得关闭资源
        consumer.close();
        session.close();
        connection.close();
        System.out.println("接收完毕");
    }
}

// 6、发送消息（生成者）, 默认是持久化的，重启之后，数据还在。
MessageProducer producer = session.createProducer(topic);

// 练习测试：先发送消息，然后关掉MQ服务。（对比测试）
message.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);
message.setJMSDeliveryMode(DeliveryMode.PERSISTENT);

// 持久化策略开启后再连接(注意即可，消息发送就持久化)
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
connection.start();

// 设置连接ID，我是谁，
connection.setClientID("icoding");

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("topic-coding-persist");
// 5、创建订阅人
TopicSubscriber subscriber = session.createDurableSubscriber(topic, "coding老师");
// 开启连接
connection.start();

// true: 先执行send方法，然后执行 commit 才是真正的放入队列。可以积压多个消息，然后一次发送
// false: send,直接进入主题，如果你关闭了事务，签收一定要设置为自动签收。
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

session.commit();

Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

public interface Session extends Runnable {
    int AUTO_ACKNOWLEDGE = 1;   // 自动签收
    int CLIENT_ACKNOWLEDGE = 2;  // 客户端确认签收
    int DUPS_OK_ACKNOWLEDGE = 3; // 自动批量进行签收
    int SESSION_TRANSACTED = 0;  // 事务提交就签收   
}