ActiveMQ(一)之初始ActiveMQ

系统管理员 2022-05-22 09:21 250阅读 0赞

# 一、ActiveMQ #

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

# 二、消息传递方式介绍 #

①.Activemq支持两种方式的消息传递:
    
        广播模式:1-n的方式,是一种发布订阅模式,像腾讯新闻那样,只要我们微信关注了腾讯新闻,那么每个人都会收到推送的新闻
    
        队列模式:1-1的方式,只能有一个消费者端消费生产者生产的数据
    ②.消息类型介绍:
    
            Activemq提供了两种消息类型:持久化和非持久化:
    
            消息生产者使用持久（persistent）传递模式发送消息的时候，Producer.send() 方法会被阻塞，直到 broker 发送一个确认消息给生产者(ProducerAck)，这个确认消息暗示broker已经成功接收到消息并把消息保存到二级存储中。这个过程通常称为同步发送。速度较慢,数据基本不会丢失.可以持久化到kahaDB(aMq默认采用kahaDB存储引擎来存储消息)或数据库中

异步发送不会在受到 broker 的确认之前一直阻塞 Producer.send 方法,速度较快,不过可能会造成数据的丢失.

消息签收方式:

AUTO\_ACKNOWLEDGE    自动确认

CLIENT\_ACKNOWLEDGE    客户端手动确认

DUPS\_OK\_ACKNOWLEDGE   自动批量确认

SESSION\_TRANSACTED    事务提交并确认

下面我以一个程序例子来展现，ActiveMQ这里我安装在win上，具体的安装教程这里就不介绍了。

1.ActiveMQ的1-1模式下的发送者

/**
    * @author JokerMqc
    * @time 2018/6/20 17:34
    * @version V1.0.0
    * @description ActiveMQ发送者测试
    */
    public class AppProducer {
        /**
         *  连接地址
         */
        private static final String URL = "tcp://localhost:61616";
        /**
         * 创建的队列名
         */
        private static final String queueName = "queue-test";
    
        public static void main(String[] args) throws JMSException {
            // 1. 获取连接工厂对象
            ConnectionFactory factory = new ActiveMQConnectionFactory(URL);
    
            // 2. 获取连接对象
            Connection connection = factory.createConnection();
    
            // 3. 启动连接
            connection.start();
    
            // 4. 创建会话 第一个参数代表的是是否启动事务 第二个参数表带的是接受的机制
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
            // 5. 创建目的地
            Destination destination = session.createQueue(queueName);
    
            // 6. 创建对应的生产者
            MessageProducer producer = session.createProducer(destination);
    
            for (int i = 0 ; i < 100; i ++){
                // 7. 创建消息对象
                TextMessage textMessage = session.createTextMessage("消息"+i);
                producer.send(textMessage);
    
                System.out.println("发送："+textMessage.getText());
            }
    
            // 7. 关闭对应的资源
            connection.close();
        }
    }

2.ActiveMQ的1-1模式下的接收者

/**
    * @author JokerMqc
    * @time 2018/6/21 9:36
    * @version V1.0.0
    * @description ActiveMQ 1-1模式下的接收者
    */
    public class AppReceiver {
        /**
         *  连接地址
         */
        private static final String URL = "tcp://localhost:61616";
        /**
         * 创建的队列名
         */
        private static final String queueName = "queue-test";
    
        public static void main(String[] args) throws JMSException {
            // 1. 获取连接工厂对象
            ConnectionFactory factory = new ActiveMQConnectionFactory(URL);
    
            // 2. 获取连接对象
            Connection connection = factory.createConnection();
    
            // 3. 启动连接
            connection.start();
    
            // 4. 创建会话 第一个参数代表的是是否启动事务 第二个参数表带的是接受的机制
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
            // 5. 创建目的地
            Destination destination = session.createQueue(queueName);
    
            // 6. 创建对应的接收者
            MessageConsumer consumer = session.createConsumer(destination);
    
            // 7. 添加对应的监听器，回调得到对应的消息
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("接收到："+textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

3.ActiveMQ的1-n模式下的发布者

/**
    * @author JokerMqc
    * @time 2018/6/20 17:34
    * @version V1.0.0
    * @description ActiveMQ主题模式下发布者测试
    */
    public class AppProducer {
        /**
         *  连接地址
         */
        private static final String URL = "tcp://localhost:61616";
        /**
         * 创建的队列名
         */
        private static final String topicName = "queue-test";
    
        public static void main(String[] args) throws JMSException {
            // 1. 获取连接工厂对象
            ConnectionFactory factory = new ActiveMQConnectionFactory(URL);
    
            // 2. 获取连接对象
            Connection connection = factory.createConnection();
    
            // 3. 启动连接
            connection.start();
    
            // 4. 创建会话 第一个参数代表的是是否启动事务 第二个参数表带的是接受的机制
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
            // 5. 创建目的地
            Destination destination = session.createTopic(topicName);
    
            // 6. 创建对应的生产者
            MessageProducer producer = session.createProducer(destination);
    
            for (int i = 0 ; i < 100; i ++){
                // 7. 创建消息对象
                TextMessage textMessage = session.createTextMessage("消息"+i);
                producer.send(textMessage);
    
                System.out.println("发送："+textMessage.getText());
            }
    
            // 7. 关闭对应的资源
            connection.close();
        }
    }

4.ActiveMQ的1-n模式下的订阅者

/**
    * @author JokerMqc
    * @time 2018/6/21 9:36
    * @version V1.0.0
    * @description ActiveMQ 1-1模式下的接收者
    */
    public class AppReceiver {
        /**
         *  连接地址
         */
        private static final String URL = "tcp://localhost:61616";
        /**
         * 创建的队列名
         */
        private static final String topicName = "queue-test";
    
        public static void main(String[] args) throws JMSException {
            // 1. 获取连接工厂对象
            ConnectionFactory factory = new ActiveMQConnectionFactory(URL);
    
            // 2. 获取连接对象
            Connection connection = factory.createConnection();
    
            // 3. 启动连接
            connection.start();
    
            // 4. 创建会话 第一个参数代表的是是否启动事务 第二个参数表带的是接受的机制
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
            // 5. 创建目的地
            Destination destination = session.createTopic(topicName);
    
            // 6. 创建对应的接收者
            MessageConsumer consumer = session.createConsumer(destination);
    
            // 7. 添加对应的监听器，回调得到对应的消息
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("接收到："+textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

注意：ActiveMQ中的1-n模式下，订阅者无法接收到发布之前的发布的消息，也就是说订阅者必须提前预定主题才能够接收到对应的消息