JMS实战——ActiveMQ

野性酷女 2022-08-22 15:20 232阅读 0赞

## 安装 ##

官网下载地址：[http://activemq.apache.org/][http_activemq.apache.org]

小编这里以5.9.0版本为例，做简单介绍。

下载之后解压到制定路径，目录结构如下：

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

## 启动 ##

直接运行bin下的activemq.bat，出现如下界面。

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

浏览器输入[http://localhost:8161][http_localhost_8161]进行访问，这只是一个安装成功的页面。一般用admin访问，在地址后加/admin,用户名和密码都是admin。

如下图：

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

## 实现PTP ##

JMS有两种消息模型：PTP和Pub/Sub。这里以第一种简单的为例，来感性的了解下。

新建Queue：

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

代码实现：

> 整个过程涉及到消息的生产者（发出消息）、消息的消费者（接收消息）、JMS服务器（负责通信的支持，这里使用的是activeMQ的支持）。

消息的生产者 Sender

public class Sender {
    
        //消息个数
        private static final int SEND_NUMBER = 5;
    
        public static void main(String[] args) {
            //初始化开始，包括连接工厂、连接、会话、消息目的、消息生产者
            ConnectionFactory connectionFactory;
            Connection connection = null;
            Session session;
            Destination destination;
            MessageProducer producer;
    
            //创建连接工厂，使用默认用户名和密码。这里tcp://localhost:61616为连接地址，当然也可以使用默认地址。
            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);
                //创建一个名称为TestQueue的消息队列
                destination = session.createQueue("TestQueue");
    
                //得到producer
                producer = session.createProducer(destination);
    
                // 设置不持久化
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
                // 构造消息
                sendMessage(session, producer);
                session.commit();
    
            } catch (JMSException e) {
    
                e.printStackTrace();
            } finally {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }
        }
    
        private static void sendMessage(Session session, MessageProducer producer) throws JMSException {
    
            for (int i = 0; i < SEND_NUMBER; i++) {
                TextMessage message=session.createTextMessage("I am a Producer"+i);
                System.out.println("发送消息："+message.getText());
                //发送给制定用户
                message.setStringProperty("username", "A");
                producer.send(message);
            }
        }
    }

> 这里采用的是发送给指定用户（username=“A”），当然，也可以不加限制，只要在同一消息队列中的消费者都可以接收。

消息的消费者 Receiver

try {
                String clientId = "003";
                connection = connectionFactory.createConnection();
                // 设置客户端id
                connection.setClientID(clientId);
                connection.start();
                // 创建会话
                session = connection.createSession(Boolean.FALSE,
                        Session.AUTO_ACKNOWLEDGE);
                // 使用同一个消息队列
                destination = session.createQueue("TestQueue");
    
                // consumer=session.createConsumer(destination);
                // 指定当前登录的标识
                consumer = session.createConsumer(destination, "username='A'");
    
                while (true) {
                    // 接收消息
                    TextMessage message = (TextMessage) consumer.receive(1000000);
                    if (null != message) {
                        System.out.println("clientID：" + connection.getClientID()
                                + " 收到消息：" + message.getText());
                    } else {
                        break;
                    }
                }
            } catch (Exception e) {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }

控制台显示：

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

浏览器中查看Queue状态：

在TestQueue队列中，5个消息入队，5个消息出队。

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

[http_activemq.apache.org]: http://activemq.apache.org/
[20160601113725581]: /images/20220722/fdb6a9ac903f42f7a89daa84b373a7d2.png
[20160601113957582]: /images/20220722/d45d306118fa43af8dfbcb30cf79c9c7.png
[http_localhost_8161]: http://localhost:8161
[20160601114208371]: /images/20220722/723b21c0cb24455284e1945d07414b17.png
[20160601114912366]: /images/20220722/1a107bd9eb784cce8f72f704ff1cbbcf.png
[20160601145147982]: /images/20220722/39c8230e74304cd080b9310b3f84019f.png
[20160601144505262]: /images/20220722/279e86a83c8d46939d1352ba37e4b036.png