ActiveMQ的学习整理（代码实现PTP，以及Pub/Sub）

（一）由于在实习过程中需要用到ActiveMQ，在网上看了很多文章，现在整理出来以防忘记。

（二）这篇文章比较适合之前没有接触过的同学，在看下面文章的过程中，建议先学习参考链接中的知识点，然后自己再参考我的代码来实现实现PTP，以及Pub/Sub，两遍下来基本能搞定ActiveMQ的基础知识。

（三）参考链接：https://www.jianshu.com/p/b547db55d168

这是关于消息中间件ActiveMQ的一个系列专题文章，将涵盖JMS、ActiveMQ的初步入门及API详细使用、两种经典的消息模式（PTP and Pub/Sub）、与Spring整合、ActiveMQ集群、监控与配置优化等。

JMS

首先来说较早以前，也就是没有JMS的那个时候，很多应用系统存在一些缺陷：

1.通信的同步性：client端发起调用后，必须等待server处理完成并返回结果后才能继续执行；

2.client 和 server 的生命周期耦合太高：client进程和server服务进程都必须可用，如果server出现问题或者网络故障，那么client端会收到异常；

3.点对点通信：client端的一次调用只能发送给某一个单独的服务对象，无法一对多；

JMS，即Java Message Service，通过面向消息中间件（MOM：Message Oriented Middleware）的方式很好的解决了上面的问题。大致的过程是这样的：发送者把消息发送给消息服务器，消息服务器将消息存放在若干队列/主题中，在合适的时候，消息服务器会将消息转发给接受者。在这个过程中，发送和接受是异步的，也就是发送无需等待，而且发送者和接受者的生命周期也没有必然关系；在pub/sub模式下，也可以完成一对多的通信，即让一个消息有多个接受者。

需要注意的是，JMS只是定义了Java访问消息中间件的接口，其实就是在包javax.jms中，你会发现这个包下除了异常定义，其他都是interface。我们可以扫一眼，比如Message：

JMS只给出接口，然后由具体的中间件去实现，比如ActiveMQ就是实现了JMS的一种Provider，还有阿里巴巴的RocketMQ（后续专题中在为大家介绍）。这些消息中间件都符合JMS规范。说起规范，自然要定义一些术语：

Provider/MessageProvider：生产者

Consumer/MessageConsumer：消费者

PTP：Point To Point，点对点通信消息模型

Pub/Sub：Publish/Subscribe，发布订阅消息模型

Queue：队列，目标类型之一，和PTP结合

Topic：主题，目标类型之一，和Pub/Sub结合

ConnectionFactory：连接工厂，JMS用它创建连接

Connnection：JMS Client到JMS Provider的连接

Destination：消息目的地，由Session创建

Session：会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是Session创建的

PTP模式实现

PTP模式特点：

1，每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中) 
2，发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列 
3，接收者在成功接收消息之后需向队列应答成功

（一）在IntelliJ IDEA中新建project，然后再建立两个项目，以及相应的类

（二）启动ActiveMQ

双击apache-activemq-5.5.1inactivemq.bat运行ActiveMQ程序。启动ActiveMQ以后，登陆：http://localhost:8161/admin/（默认账号密码为admin），创建一个Queue，命名为FirstQueue。

详细过程参考：https://www.cnblogs.com/xwdreamer/archive/2012/02/21/2360818.html

（三）sender与receiver的代码

sender：

 1 import org.apache.activemq.ActiveMQConnection;
 2 import org.apache.activemq.ActiveMQConnectionFactory;
 3 
 4 import javax.jms.*;
 5 import java.util.Date;
 6 
 7 public class sender {
 8     private static final int SEND_NUMBER = 5;
 9 
10     public static void main(String[] args) {
11         // ConnectionFactory ：连接工厂，JMS 用它创建连接
12         ConnectionFactory connectionFactory;
13         // Connection ：JMS 客户端到JMS Provider 的连接
14         Connection connection = null;
15         // Session： 一个发送或接收消息的线程
16         Session session;
17         // Destination ：消息的目的地;消息发送给谁.
18         Destination destination;
19         // MessageProducer：消息发送者
20         MessageProducer producer;
21 
22         // TextMessage message;
23         // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
24         connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616");
25         try {
26             // 构造从工厂得到连接对象
27             connection = connectionFactory.createConnection();
28             // 启动
29             connection.start();
30             // 获取操作连接
31             session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
32             // 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
33             destination = session.createQueue("FirstQueue");
34             // 得到消息生成者【发送者】
35             producer = session.createProducer(destination);
36             // 设置不持久化，此处学习，实际根据项目决定
37             producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
38             // 构造消息，此处写死，项目就是参数，或者方法获取
39             sendMessage(session, producer);
40             session.commit();
41         } catch (Exception e) {
42             e.printStackTrace();
43         } finally {
44             try {
45                 if (null != connection) {
46                     connection.close();
47                 }
48             } catch (Throwable ignore) {
49             }
50         }
51     }
52 
53     public static void sendMessage(Session session, MessageProducer producer) throws Exception {
54         for (int i = 1; i <= SEND_NUMBER; i++) {
55             Date date = new Date();
56             TextMessage message = session.createTextMessage(date + "  ActiveMQ 发送的消息:" + i);
57             // 发送消息到目的地方
58             System.out.println("发送消息：     " + date + "  ActiveMQ 发送的消息  " + i);
59             producer.send(message);
60         }
61     }
62 }
View Code

receiver：

import org.apache.activemq.*;
import javax.jms.*;

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) {
            }
        }
    }
}
View Code

（四）调试效果

Pub/Sub模式

Pub/Sub模式特点 

1，每个消息可以有多个消费者 
2，发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。 
3，为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。

 （一）新建项目

（二）show the code

TopicProducer：
 1 package pub_sub;
 2 
 3 import org.apache.activemq.*;
 4 import javax.jms.*;
 5 import java.util.Date;
 6 
 7 public class TopicProducer {
 8     private static final int SEND_NUMBER = 5;
 9     public static void main(String[] args) {
10         // ConnectionFactory ：连接工厂，JMS 用它创建连接
11         ConnectionFactory connectionFactory;
12         // Connection ：JMS 客户端到JMS Provider 的连接
13         Connection connection = null;
14         // Session： 一个发送或接收消息的线程
15         Session session;
16         // Destination ：消息的目的地;消息发送给谁.
17         Destination destination;
18         // MessageProducer：消息发送者
19         MessageProducer producer;
20         // TextMessage message;
21         // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
22         connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616");
23 
24         try {
25             // 构造从工厂得到连接对象
26             connection = connectionFactory.createConnection();
27             // 启动
28             connection.start();
29             // 获取操作连接
30             session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
31 
32             // 获取session注意参数值是一个服务器的Topic，须在在ActiveMq的console配置
33             destination = session.createTopic("FirstTopic");
34             producer = session.createProducer(destination);
35             // 设置不持久化，此处学习，实际根据项目决定
36             producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
37             // 构造消息，此处写死，项目就是参数，或者方法获取
38             sendMessage(session, producer);
39             session.commit();
40 
41         } catch (Exception e) {
42             e.printStackTrace();
43         }
44         finally {
45             try {
46                 if (null != connection) {
47                     connection.close();
48                 }
49             } catch (Throwable ignore) {
50             }
51         }
52     }
53 
54     public static void sendMessage(Session session, MessageProducer producer) throws Exception {
55         for (int i = 1; i <= SEND_NUMBER; i++) {
56             Date date = new Date();
57             TextMessage message = session.createTextMessage(date + "  ActiveMQ 发送的消息:" + i);
58             // 发送消息到目的地方
59             System.out.println("发送消息：     " + date + "  ActiveMQ 发送的消息  " + i);
60             producer.send(message);
61         }
62     }
63 }
View Code
TopicConsumer：
 1 package receiver;
 2 
 3 import org.apache.activemq.*;
 4 import javax.jms.*;
 5 
 6 public class TopicConsumer{
 7 
 8     public static void main(String[] args) {
 9         // ConnectionFactory ：连接工厂，JMS 用它创建连接
10         ConnectionFactory connectionFactory;
11         // Connection ：JMS 客户端到JMS Provider 的连接
12         Connection connection = null;
13         // Session： 一个发送或接收消息的线程
14         Session session;
15         // Destination ：消息的目的地;消息发送给谁.
16         Destination destination;
17         // 消费者，消息接收者
18         MessageConsumer consumer;
19         connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
20 
21         try {
22             // 构造从工厂得到连接对象
23             connection = connectionFactory.createConnection();
24             // 启动
25             connection.start();
26             // 获取操作连接
27             session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
28             // 获取session注意参数值是一个服务器的Topic，须在在ActiveMq的console配置
29             destination = session.createTopic("FirstTopic");
30             consumer = session.createConsumer(destination);
31 //            while (true) {
32 //                //设置接收者接收消息的时间，为了便于测试，这里谁定为100s
33 //                TextMessage message = (TextMessage) consumer.receive(100000);
34 //                if (null != message) {
35 //                    System.out.println("收到消息:" + message.getText());
36 //                } else {
37 //                    break;
38 //                }
39 //            }
40             while(true){
41                 consumer.setMessageListener(new listern());
42             }
43         } catch (Exception e) {
44             e.printStackTrace();
45         }
46         finally {
47             try {
48                 if (null != connection){
49                     connection.close();
50                 }
51             } catch (JMSException e) {
52                 e.printStackTrace();
53             }
54         }
55     }
56 }
View Code
listern:
 1 package receiver;
 2 
 3 import javax.jms.JMSException;
 4 import javax.jms.Message;
 5 import javax.jms.MessageListener;
 6 import javax.jms.TextMessage;
 7 
 8 public class listern implements MessageListener {
 9     @Override
10     public void onMessage(Message message) {
11         try {
12             System.out.println("收到的消息:"+((TextMessage)message).getText());
13         } catch (JMSException e) {
14             // TODO Auto-generated catch block
15             e.printStackTrace();
16         }
17     }
18 }
View Code

（三）注意事项

1. 运行代码的时候要先运行订阅者代码，然后运行发布者代码。

2. Consumer中注释代码为不添加Listern的实现。

说明：所有内容仅做学习记录
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/jayinnn/p/9598030.html