ActiveMQ (一) 初试

阳光穿透心脏的1/2处 2022-05-29 00:38 212阅读 0赞

## 一.什么是mq? ##

ActiveMQ是一种开源的，实现了JMS1.1规范的，面向消息(MOM)的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。

*JMS的全称是Java Message Service，即Java消息服务。它主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息，而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息，并进行发送，对应的消费者在接收到对应的消息后去完成对应的业务逻辑。对于消息的传递有两种类型，一种是点对点的，即一个生产者和一个消费者一一对应；另一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。*

## 二.为什么需要mq? ##

1.  解耦：每个成员不必受其他成员影响，可以更独立自主，只通过一个简单的容器来联系。更适合作为多个应用之间的松耦合的接口。
2.  提速：异步处理，非核心流程异步化，提高系统响应性能，生产者不需要从消费者处获得反馈。
3.  广播：生产者发布一条消息，可以有多个消费者接收。
4.  流量削峰与流控：当上下游系统处理能力存在差距的时候，利用消息队列做一个通用的“漏斗”。在下游有能力处理的时候，再进行分发。

## 三、体验activemq ##

官网有下载，我这里下载的windows版本的。

![这里写图片描述][SouthEast]

注意有个类似tomcat目录下的webapps目录，ActiveMQ自带Jetty提供Web管控台

进入bin目录下的32或者64目录，运行acrivemq.bat，访问[http://localhost:8161/][http_localhost_8161]

![这里写图片描述][SouthEast 1]

登录的用户名和密码在conf/jetty-realm.properties 里面设置，登陆以后如下：

![这里写图片描述][SouthEast 2]

写一个简单的demo：需要的jar包

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-core</artifactId>
        <version>5.7.0</version>
    </dependency>

写一个生产者：

import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    /** * @author susq * @since 2018-03-13-20:16 */
    public class ProducerApp { 
        private static final Logger LOGGER = LoggerFactory.getLogger(ProducerApp.class);
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
        private static final String SUBJECT = "my.queue";
    
        public static void main(String[] args) throws JMSException {
            // 初始化连接工厂
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
            // 获得连接
            Connection connection = connectionFactory.createConnection();
            // 连接启动
            connection.start();
            // 创建session, 此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建队列
            Destination destination = session.createQueue(SUBJECT);
            // createTopic方法用来创建topic
            //session.createTopic("topic");
            // 通过session可以创建消息的生产者
            MessageProducer producer = session.createProducer(destination);
            for (int i = 0; i < 10; i++) {
                // 初始化一个mq消息
                TextMessage message = session.createTextMessage("hello susq" + i);
                // 发送消息
                producer.send(message);
                LOGGER.debug("send message {}", i);
            }
            connection.close();
        }
    }

运行生产者程序，到mq管理页面进入queue,可以看到我们创建的队列 my.queue， 还有发送的10条消息，等待消费。此时没有消费者。

*Messages Enqueued：表示生产了多少条消息*  
*Messages Dequeued：表示消费了多少条消息*  
*Number Of Consumers：表示在该队列上还有多少消费者在等待接受消息*  
*Number Of Pending Messages：表示还有多少条消息没有被消费，实际上是表示消息的积压程度*

![这里写图片描述][SouthEast 3]

写一个消费者：

import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    /** * @author susq * @since 2018-03-13-20:29 */
    public class ConsumerApp implements MessageListener { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerApp.class);
        private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
        private static final String SUBJECT = "my.queue";
    
        public static void main(String[] args) throws JMSException {
            // 初始化连接工厂
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
            // 创建mq连接
            Connection connection = connectionFactory.createConnection();
            // 启动连接
            connection.start();
            // 通过会话创建目标
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 通过会话创建目标
            Destination destination = session.createQueue(SUBJECT);
            // 创建mq的消息消费者
            MessageConsumer consumer = session.createConsumer(destination);
            // 初始化MessageListener
            ConsumerApp me = new ConsumerApp();
            // 给消费者设定监听对象
            consumer.setMessageListener(me);
        }
    
        public void onMessage(Message message) {
            TextMessage textMessage = (TextMessage) message;
            try {
                LOGGER.info("get message:" + textMessage.getText());
            } catch (JMSException e) {
                LOGGER.error("error {}", e);
            }
        }
    }

消费者需要实现MessageListener接口，此接口只有一个`onMessage(Message message)`需要在此方法中做消息的处理。  
运行后，可以看到有一个消费者，Number Of Consumers 为 1，并且刚才发送的消息已经全部被消费。这里我运行额两次生产者，所以有20条消息。

![这里写图片描述][SouthEast 4]

[SouthEast]: /images/20220529/74dc724d1bee4a50bc85b85aa7bbc2a3.png
[http_localhost_8161]: http://localhost:8161/
[SouthEast 1]: /images/20220529/14f5996eabc94945b671e39cf8c401cb.png
[SouthEast 2]: /images/20220529/776f2d5e39bf4324bc1c5995578bad71.png
[SouthEast 3]: /images/20220529/f83aed1beaab4928a31caa0d6b8f49a9.png
[SouthEast 4]: /images/20220529/a95bac5441c349e3812752be67fa5e26.png