第一个activemq 示例

太过爱你忘了你带给我的痛 2022-06-15 05:16 189阅读 0赞

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。\-- 来自度娘

以上我们可以得知，jms 是一种规范，而activemq 是jms规范的一个实现。

JMS 通讯方式通常有两种，一种点对点消息传递； 一种发布/订阅消息传递。 这里介绍一下点对点消息传递的示例。

activemq 的下载可以去官网 [http://activemq.apache.org/][http_activemq.apache.org] 下载，下载后解压到本地，启动bin 目录下面的activemq.bat 文件即可

奇怪的是，我本地的activemq.bat 一直无法直接启动起来，窗口闪退。咨询朋友后，找到启动方法。

在windows系统下，进命令行模式，到bin目录下输入  activemq start 来启动

![SouthEast][]

待出现如下日志时，便告启动成功

![SouthEast 1][]

启动成功后，在 [http://localhost:8161/admin/queues.jsp][http_localhost_8161_admin_queues.jsp] 管理页面的 Queue Name 输入框输入“HelloWorld” ，并创建Queue

![SouthEast 2][]

测试代码包含两个类，JMSProducer.java  和 JMSConsumer.java , 实现如下：

JMSProducer

package com.zhangqi.jms;
    
    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;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class JMSProducer {
    
    	// 默认连接用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        // 默认连接密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        // 默认连接地址
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
        // 发送的消息数量
        private static final int SENDNUM = 100;
    
        public static void main(String[] args) {
            //连接工厂
            ConnectionFactory connectionFactory;
            //连接
            Connection connection = null;
            //会话 接受或者发送消息的线程
            Session session;
            //消息的目的地
            Destination destination;
            //消息生产者
            MessageProducer messageProducer;
            //实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
    
            try {
                //通过连接工厂获取连接
                connection = connectionFactory.createConnection();
                //启动连接
                connection.start();
                //创建session
                session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
                //创建一个名称为HelloWorld的消息队列
                destination = session.createQueue("HelloWorld");
                //创建消息生产者
                messageProducer = session.createProducer(destination);
                //发送消息
                sendMessage(session, messageProducer);
    
                session.commit();
            } catch (Exception e) {
                e.printStackTrace();
            }finally{
                if(connection != null){
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
        
        /**
         * 发送消息
         * @param session
         * @param messageProducer  消息生产者
         * @throws Exception
         */
        public static void sendMessage(Session session,MessageProducer messageProducer) throws Exception{
            for (int i = 0; i < JMSProducer.SENDNUM; i++) {
                //创建一条文本消息 
                TextMessage message = session.createTextMessage("ActiveMQ " + i + "@#$");
                System.out.println("发送消息：Activemq 发送消息" + i);
                //通过消息生产者发出消息 
                messageProducer.send(message);
            }
        }
    }

JMSConsumer

package com.zhangqi.jms;
    
    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.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class JMSConsumer {
    
    	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址
    
        public static void main(String[] args) {
            ConnectionFactory connectionFactory;//连接工厂
            Connection connection = null;//连接
    
            Session session;//会话 接受或者发送消息的线程
            Destination destination;//消息的目的地
    
            MessageConsumer messageConsumer;//消息的消费者
    
            //实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
    
            try {
                //通过连接工厂获取连接
                connection = connectionFactory.createConnection();
                //启动连接
                connection.start();
                //创建session
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                //创建一个连接HelloWorld的消息队列
                destination = session.createQueue("HelloWorld");
                //创建消息消费者
                messageConsumer = session.createConsumer(destination);
    
                while (true) {
                    TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
                    if(textMessage != null){
                        System.out.println("收到的消息:" + textMessage.getText());
                    }else {
                        break;
                    }
                }
    
            } catch (JMSException e) {
                e.printStackTrace();
            }
    
        }
    }

测试如下：

运行JMSProducer，可以看到输出

![SouthEast 3][]

同时在activemq的管理页面可以看到

![SouthEast 4][]

Queue几列含义如下

Number Of Consumers  **消费者端的消费者数量**，我们这里是点对点，故消费者数量为1  
Number Of Pending Messages**等待消费的消息**，这个是当前未出队列的数量，可以理解为总接收数-总出队列数   
Messages Enqueued**进入队列的消息**  进入队列的总数量,包括出队列的，这个数量只增不减   
Messages Dequeued**出了队列的消息**  可以理解为是消费这消费掉的数量

待运行JMSConsumer后，可以看到输出

![SouthEast 5][]

同时 activemq管理页面

![SouthEast 6][]

这时可以看到，等待消费的消息为0，说明前期的消息都已经推送到消费者端，且进入队列消息和出队列消息均为100，说明无等待消费的消息。

好了，这个示例就先写到这里了，初次学习，若有疏漏部分请批评指正。谢谢。

[http_activemq.apache.org]: http://activemq.apache.org/
[SouthEast]: /images/20220615/de4ca902599549ea8e7fa3b159f93622.png
[SouthEast 1]: /images/20220615/7504f02fe92b47a0958c8bae9aa1ff11.png
[http_localhost_8161_admin_queues.jsp]: http://localhost:8161/admin/queues.jsp
[SouthEast 2]: /images/20220615/7e4c78b59d2c46e1b48e1af7ba93817a.png
[SouthEast 3]: /images/20220615/3186f7dbd49746d1bb46df956674353d.png
[SouthEast 4]: /images/20220615/42fe77e4a08642f5bbfd3f02f37a03a4.png
[SouthEast 5]: /images/20220615/26d2e1b4b58d4e6d90bd1452cf3011d7.png
[SouthEast 6]: /images/20220615/3ed078b3e6c348489e6d548ce94f1727.png