学习ActiveMQ(二):点对点(队列)模式消息演示

时间:2019-04-20
本文章向大家介绍学习ActiveMQ(二):点对点(队列)模式消息演示,主要包括学习ActiveMQ(二):点对点(队列)模式消息演示使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。

  一:介绍

  点对点的消息发送方式主要建立在 消息(Message ),队列(Queue),发送者(Sender),消费者(receiver)上,Queue 存贮消息,Sender 发送消息,receive接收消息.具体点就是Sender Client通过Queue发送message ,而 receiver Client从Queue中接收消息。消息发送客户端与接收客户端没有时间上的依赖,发送客户端可以在任何时刻发送信息到Queue,而不需要知道接收客户端是不是在运行。
  二:了解jms编码接口之间的关系

  可以看到我们需要先创建连接工厂,再通过连接工厂创建连接,在通过连接创建session,在通过session创建生产者、消费者和消息。destination为目的源,下面代码有具体体现。

  三:创建实例

  1.打开IDEA,创建一个ActiveMQ的maven项目,如下图:

  2.自己新创建两个java文件,appConsumer消费者类和appProducer生产者类,项目结构如下图:

   3.生产者代码如下:

 1 package com.liu.jms;
 2 
 3 import org.apache.activemq.ActiveMQConnectionFactory;
 4 
 5 import javax.jms.*;
 6 
 7 public class appProducer {
 8 
 9     private static final String url = "tcp://127.0.0.1:61616";//actvemq的服务器tcp连接方式
10     private static final String queueName = "queue-test";//定义队列的名称
11 
12     public static void main(String[] args) throws  JMSException {
13         //1.创建connectionFactory
14         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
15         //2.创建connection
16         Connection connection = connectionFactory.createConnection();
17         //3.启动连接
18         connection.start();
19         //4.创建session
20         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
21         //5.创建destination
22         Destination destination = session.createQueue(queueName);
23         //6.创建生产者
24         MessageProducer producer = session.createProducer(destination);
25 
26         for (int i = 0; i < 100; i++) {
27 
28             TextMessage textMessage = session.createTextMessage("test" + i);
29             //7.发送消息
30             producer.send(textMessage);
31 
32             System.out.println("发送消息" + textMessage.getText());
33 
34         }
35         //8.关闭连接
36         connection.close();
37     }
38 }

  如代码所示,通过tcp方式连接了服务端,(别忘了启动服务端的服务)。然后创建了一个生产者,这个生产者绑定了一个以名为queueName队列为目的源,代表着这个生产者的消息会发到这个消息队列上面去。然后通过for循环发送了一百个消息。

  4.消费者代码如下:

package com.liu.jms;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class appConsumer {

    private static final String url = "tcp://127.0.0.1:61616";
    private static final String queueName = "queue-test";

    public static void main(String[] args) throws  JMSException {
        //1.创建connectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        //2.创建connection
        Connection connection = connectionFactory.createConnection();
        //3.启动连接
        connection.start();
        //4.创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.创建destination
        Destination destination = session.createQueue(queueName);
        //6.创建消费者
        MessageConsumer consumer = session.createConsumer(destination);
        //7.创建一个监听器
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {

                TextMessage textMessage = (TextMessage)message;
                try {
                    System.out.println("接收到的消息:" + textMessage.getText());

                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //8.关闭连接(监听器是异步的还没有监听到消息的时候,就关闭连接了)
        //connection.close();
    }
}

  如代码所示:消费者和生产者不同的是,消费需要建立一个监听器,来监听以名为queueName的队列上是否有了消息,有消息就会接受,然后通过onMessage方法对消息进行处理。

   5.测试

  首先启动消费者这个java类,观察控制台,如下图:

  接着启动生产者的java类,观察控制台,如下图:生产了一百条消息。

  此时切换至消费的控制台,观察控制台,如下图:已经打印出了一百条消息了,说明消费者已经接受到全部一百条消息。

   6.打开activemq的控制台查看Queues:(http://127.0.0.1:8161/admin/queues.jsp)如下图所示:队列有一个名字是我们设置的queue-test,消费者也有一个就是我们创建的那个消费者类,队列中有一百条消息,被移除了一百条,也就是上面所说的,消费者接收到了这100条全部的消息。

  通过这个简单的小demo我们已经实现了点对点的通信方式。下一节学习订阅发布模式。