java 消息机制 ActiveMQ入门实例

た入场券 2022-06-16 08:42 175阅读 0赞

> 1.下载ActiveMQ  
> 去官方网站下载：[http://activemq.apache.org/][http_activemq.apache.org]  
> 我下载的是 apache-activemq-5.14.5 Release版

> 2.运行ActiveMQ  
> 解压缩apache-activemq-5.14.5-bin.zip，然后双击apache-activemq-5.14.5-bin/bin/win64/activemq.bat运行ActiveMQ程序。  
> 启动ActiveMQ以后，登陆：[http://localhost:8161/admin/][http_localhost_8161_admin]，创建一个Queue，命名为FirstQueue。  
> PS：帐号 admin 密码 admin  
> ![这里写图片描述][SouthEast]

> 3.创建IntelliJ IDEA项目并运行  
> 创建java project：activemq，新建lib文件夹  
> 打开activemq\\lib目录  
> 拷贝  
> activemq-broker-5.14.5.jar  
> activemq-client-5.14.5.jar  
> geronimo-j2ee-management\_1.1\_spec-1.0.1.jar  
> geronimo-jms\_1.1\_spec-1.1.1.jar  
> slf4j-api-1.7.13.jar
> 
> PS:还有一个包，不再apache-activemq-5.14.5-bin.zip里，要下载一个包activemq-all，地址：  
> [https://repository.apache.org/content/repositories/releases/org/apache/activemq/activemq-all/][https_repository.apache.org_content_repositories_releases_org_apache_activemq_activemq-all]  
> 找到对应的版本，我这里用的是5.14.5版本，点击进去，下载如图  
> ![这里写图片描述][SouthEast 1]
> 
> 这6个jar文件到lib文件夹中,并Build Path->Add as Library 如图  
> ![这里写图片描述][SouthEast 2]
> 
> 结构如图:  
> ![这里写图片描述][SouthEast 3]

Sender.java

package com.activemq;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Sender {
        
        private static final int SEND_NUMBER = 5;
    
        public static void main(String[] args) {
            // ConnectionFactory ：连接工厂，JMS 用它创建连接
            ConnectionFactory connectionFactory; // Connection ：JMS 客户端到JMS
            // Provider 的连接
            Connection connection = null; // Session： 一个发送或接收消息的线程
            Session session; // Destination ：消息的目的地;消息发送给谁.
            Destination destination; // MessageProducer：消息发送者
            MessageProducer producer; // TextMessage message;
            // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
            connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnection.DEFAULT_USER,
                    ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
            try { // 构造从工厂得到连接对象
                connection = connectionFactory.createConnection();
                // 启动
                connection.start();
                // 获取操作连接
                session = connection.createSession(Boolean.TRUE,
                        Session.AUTO_ACKNOWLEDGE);
                // 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
                destination = session.createQueue("FirstQueue");
                // 得到消息生成者【发送者】
                producer = session.createProducer(destination);
                // 设置不持久化，此处学习，实际根据项目决定
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                // 构造消息，此处写死，项目就是参数，或者方法获取
                sendMessage(session, producer);
                session.commit();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }
        }
    
        public static void sendMessage(Session session, MessageProducer producer)
                throws Exception {
            for (int i = 1; i <= SEND_NUMBER; i++) {
                TextMessage message = session.createTextMessage("ActiveMq 发送的消息"
                        + i);
                // 发送消息到目的地方
    
                System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
                producer.send(message);
            }
        }
    }

Receiver.java

package com.activemq;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Receiver {
        
        public static void main(String[] args) {
            // ConnectionFactory ：连接工厂，JMS 用它创建连接
            ConnectionFactory connectionFactory;
            // Connection ：JMS 客户端到JMS Provider 的连接
            Connection connection = null;
            // Session： 一个发送或接收消息的线程
            Session session;
            // Destination ：消息的目的地;消息发送给谁.
            Destination destination;
            // 消费者，消息接收者
            MessageConsumer consumer;
            connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnection.DEFAULT_USER,
                    ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
            try {
                // 构造从工厂得到连接对象
                connection = connectionFactory.createConnection();
                // 启动
                connection.start();
                // 获取操作连接
                session = connection.createSession(Boolean.FALSE,
                        Session.AUTO_ACKNOWLEDGE);
                // 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
                destination = session.createQueue("FirstQueue");
                consumer = session.createConsumer(destination);
                while (true) {
                    // 设置接收者接收消息的时间，为了便于测试，这里谁定为100s
                    TextMessage message = (TextMessage) consumer.receive(100000);
                    if (null != message) {
                        System.out.println("收到消息" + message.getText());
                    } else {
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }
        }
    }

> 5.测试过程  
> 先运行:Receiver.java  
> 再运行:Sender.java  
> 选中要运行的java代码，鼠标右键点击如图：  
> ![这里写图片描述][SouthEast 4]
> 
> 可以看到结果  
> Sender运行后:  
> 发送消息：ActiveMq 发送的消息1  
> 发送消息：ActiveMq 发送的消息2  
> 发送消息：ActiveMq 发送的消息3  
> 发送消息：ActiveMq 发送的消息4  
> 发送消息：ActiveMq 发送的消息5
> 
> Receiver运行后:  
> 收到消息ActiveMq 发送的消息1  
> 收到消息ActiveMq 发送的消息2  
> 收到消息ActiveMq 发送的消息3  
> 收到消息ActiveMq 发送的消息4  
> 收到消息ActiveMq 发送的消息5

END…

[http_activemq.apache.org]: http://activemq.apache.org/
[http_localhost_8161_admin]: http://localhost:8161/admin/
[SouthEast]: /images/20220616/0d9a7500e29f4040841853112ff5e43d.png
[https_repository.apache.org_content_repositories_releases_org_apache_activemq_activemq-all]: https://repository.apache.org/content/repositories/releases/org/apache/activemq/activemq-all/
[SouthEast 1]: /images/20220616/61ea4b2b08cd41f6b3c10eb0cd632a18.png
[SouthEast 2]: /images/20220616/cdcfac26b374444197d4fc10cbae0fe6.png
[SouthEast 3]: /images/20220616/3446507d213d47a289cc76401c36035a.png
[SouthEast 4]: /images/20220616/6d5ff068b29b43c48b86da14e8b5848e.png