ActiveMQ实战(二) 基本使用和功能介绍

电玩女神 2023-08-17 16:28 130阅读 0赞

**目录**

一.基本的使用

1.消息队列的发送

1.1 建立maven工程

1.2 书写发送方代码

1.3 生产者发送消息

2. 消息队列的接收

2.1 书写接收方代码

二.功能的介绍（预了解）

2.1 同步

2.2 异步

2.3 点对点

2.4 发布/订阅（publish/subscribe）（pub/sub）

--------------------

# 一.基本的使用 #

## 1.消息队列的发送 ##

### 1.1 建立maven工程 ###

如下图，新建maven工程（新建maven工程之前需要确保eclipse集成maven，可百度eclipse安装maven）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70][]

默认当前工作空间

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 1][]

点击quickstart

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 2][]

设置3个必要参数，组ID，骨架ID，包名：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 3][]

一个新的maven项目：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 4][]

修改pom.xml，新增依赖

<dependency>
        	<groupId>org.apache.activemq</groupId>
        	<artifactId>activemq-all</artifactId>
        	<version>5.9.0</version>
        </dependency>
        <dependency>
        	<groupId>org.apache.xbean</groupId>
        	<artifactId>xbean-spring</artifactId>
        	<version>3.16</version>
        </dependency>

### 1.2 书写发送方代码 ###

在jmstest中新建queue包，文件结构可以自定义，这里没有什么要求，就是普通的java代码。

![2019092819292448.png][]

新建发送类：QueueSender.java

import javax.jms.Connection;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 	实践生产者消费者模式，消息队列
     * @author Administrator
     *
     */
    public class QueueSender {
    	public static void main(String[] args) throws JMSException {
    		//建立工厂
    		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://192.168.239.129:61616");
    		//工厂模式获取连接
    		Connection connection = activeMQConnectionFactory.createConnection();
    		//连接
    		connection.start();
    		//通过连接获取会话
    		Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    		//Destination destination= session.createQueue("Hello ActiveMQ");
    		//通过会话获取队列，设置队列名
    		Queue createQueue = session.createQueue("Hello ActiveMQ");
    		//通过会话获取生产者
    		MessageProducer producer = session.createProducer(createQueue);
    		//通过会话设置需要发送的文本
    		TextMessage textMessage = session.createTextMessage("anakki");
    		//通过生产者发送文本
    		producer.send(textMessage);
    		//提交本次会话
    		session.commit();
    		//会话关闭
    		session.close();
    		//连接关闭
    		connection.close();
    	}
    }

### 1.3 生产者发送消息 ###

右键类名运行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 5][]

查看结果：我们在实战（一）中登录的管理页面查看，消息已经被队列接受了。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 6][]

## 2. 消息队列的接收 ##

### 2.1 书写接收方代码 ###

（我们可以看出发送方和接收方是完全**解耦合**的，因为此时我们接受方的代码都还未写。可是发送方已经可以发送消息了。注意这里提到的消息队列重要的特点，解耦合）

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    /**
     * 	实践生产者消费者模式，消息队列
     * @author Administrator
     *  消费者
     */
    public class QueueReceiver {
    	public static void main(String[] args) throws JMSException {
    		//建立工厂
    		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://192.168.239.129:61616");
    		//工厂模式获取连接
    		Connection connection = activeMQConnectionFactory.createConnection();
    		//连接
    		connection.start();
    		//通过连接获取会话
    		Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    		//Destination destination= session.createQueue("Hello ActiveMQ");
    		//通过会话获取队列，设置队列名
    		Queue createQueue = session.createQueue("Hello ActiveMQ");
    		
    		
    		//通过会话获取消费者
    		MessageConsumer consumer = session.createConsumer(createQueue);//A
    		//消费者接受信息
    		TextMessage message = (TextMessage)consumer.receive();//B
    		//提交本次会话
    		session.commit();
    		System.out.println(message.getText());//C
    		//会话关闭
    		session.close();
    		//连接关闭
    		connection.close();
    	}
    }

运行后可以看见接收到的结果

![20190929025024565.png][]

管理页面

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 7][]

由于我测试了多次 所以发送方一共发送了8条消息，所以入队8次enqueued8次，消费者消费8次，dequeued8次，当然消费者如果只消费了6次，那么上面将是8和6，接收和发送都是异步的。当我们队列里没有消息时，如果我们去运行消费者。那么消费者会一直等待队列里有消息，并消费掉它。这里会被Number of Consumers记录。这里表示有一个消费者在等待接收并消费队列中的消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 8][]

当然也有限定等待时间的receive方法，可以看见下图还有不阻塞的receive方法。

那我们可以得出消息队列在目前有三种接收方式：

receive（）有消息接收，没消息阻塞

receive（long timeout）有消息接收，无消息阻塞timeout毫秒数，会返回一个null给TextMessage message

receiveNoWait（）立即接收不等待，这个之后会介绍。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 9][]

我们这里已经走完了消息的发送和接收过程。

# **二.功能的介绍（预了解）** #

## **2.1 同步** ##

receive方法显式的阻塞的去接收队列消息。同步，这里可以去联想ajax中把async设为false的情况，就是此线程需要阻塞到得到消息才继续运行。

## **2.2 异步** ##

注册消息监听器。当有消息到达时，再接收消息。不会阻塞消费者。之后会实践**listener（监听器）**的代码

## **2.3 点对点** ##

每条消息只有一个消费者。

消费者取出消息后，消息就无效了。

消费者的运行和生产者的运行没有时间上的要求。即使没有生产者生产消息。当队列中有消息时消费者也能在队列中取到消息，反之亦然。

## **2.4 发布/订阅（publish/subscribe）（pub/sub）** ##

在上面点对点的消息传递中，目的地为**队列**（queue），而发布订阅消息传递中，目的地称为**主题**（topic）

有时间相关性，消费者这能收到订阅主题之后的消息。之前的消息，消费者无法接受到生产者所发布的消息。

但是有一种情况，topic又分持久订阅，和**非持久订阅**。

当为**持久订阅**时：消费者离线后，生产者发布的消息，消费者上线后也能收到。

以上加粗的关键字需要熟悉一下，在这里让我们忘掉activemq，很多其他消息队列依旧有以上功能上的要求已经对应不同的实现。因此我们需要学习消息队列的基本运行模型和JMS规范。下一节我们继续了解一下JMS相关的知识内容。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70]: /images/20230810/0faff35da0cb43b4aed65ece45ff9d53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 1]: /images/20230810/4d3a509e029f45d8bab1d78f29896ed1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 2]: /images/20230810/d2442d1445fc4d07959650d6a13aad84.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 3]: /images/20230810/5589e09669004edabf56b09120ecf696.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 4]: /images/20230810/66e8d416a16243a49d0e0579ab01db6f.png
[2019092819292448.png]: /images/20230810/2a62661b0f4b410abdf4ead13b515239.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 5]: /images/20230810/3a5b84dba5fc4cc8a7d39aec2100519e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 6]: /images/20230810/dfefa53ec7254ab3837d6401ccf89bf6.png
[20190929025024565.png]: /images/20230810/e310acecf92e4233a3be58ff6d4b0042.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 7]: /images/20230810/b61201fa79b04b068ad6b219822026dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 8]: /images/20230810/756d2c8236be48dcb85fd9f644e59bdb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 9]: /images/20230810/4cab30b5265c424fbe5de9874fd662fa.png