當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

arouter跨module传递消息_消息队列中间件（二）使用 ActiveMQ

發(fā)布時(shí)間：2025/4/5 编程问答 34 豆豆

生活随笔收集整理的這篇文章主要介紹了 arouter跨module传递消息_消息队列中间件（二）使用 ActiveMQ 小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

ActiveMQ 介紹

Active MQ 是由 Apache 出品的一款流行的功能強(qiáng)大的開源消息中間件，它速度快，支持跨語言的客戶端，具有易于使用的企業(yè)集成模式和許多的高級功能，同時(shí)完全支持?JSM1.1?和 J2EE1.4 。

官方下載地址：
?http://activemq.apache.org/download.html??
官方安裝教程：
?http://activemq.apache.org/getting-started.html??
默認(rèn)管理頁面：
http://127.0.0.1:8161/admin/??
默認(rèn)用戶名和密碼為
admin / admin。
conf / jetty-real.properties 您可以在此文件中進(jìn)行配置。??
默認(rèn)服務(wù)端口：61616??

ActiveMQ 特點(diǎn)

支持Java，C，C ++，C＃，Ruby，Perl，Python，PHP等各種跨語言客戶端和協(xié)議，如 OpenWire , Stomp , AMQP , MQTT.
完全支持JMS 1.1和 J2EE 1.4，支持瞬態(tài)，持久，事務(wù)和XA消息傳遞。
對 Spring 框架的支持以便ActiveMQ可以輕松嵌入到Spring應(yīng)用程序中。
通過了常見的 J2EE 服務(wù)器測試，如 TomEE，Geronimo，JBoss，GlassFish 和 WebLogic 。
連接方式的多樣化，ActiveMQ 提供了多種連接模式，例如 in-VM、TCP、SSL、NIO、UDP、多播、JGroups、JXTA。
可以通過使用 JDBC 和 journal 實(shí)現(xiàn)消息的快速持久化。
專為高性能群集，客戶端 - 服務(wù)器，點(diǎn)對點(diǎn)通信而設(shè)計(jì)。
提供與語言無關(guān)的 REST API。
支持 Ajax 方式調(diào)用 ActiveMQ。
ActiveMQ 可以輕松地與 CXF、Axis 等 Web Service 技術(shù)整合，以提供可靠的消息傳遞。
可用作為內(nèi)存中的 JMS 提供者，非常適合 JMS 單元測試。

ActiveMQ 消息

點(diǎn)對點(diǎn)隊(duì)列模式
消息到達(dá)消息系統(tǒng)，被保留在消息隊(duì)列中，然后由一個(gè)或者多個(gè)消費(fèi)者消費(fèi)隊(duì)列中的消息，一個(gè)消息只能被一個(gè)消費(fèi)者消費(fèi)，然后就會(huì)被移除。例如訂單處理系統(tǒng)。

發(fā)布-訂閱模式
消息發(fā)送時(shí)指定主題(或者說通道)，消息被保留在指定的主題中，消費(fèi)者可以訂閱多個(gè)主題，并使用主題中的所有的消息，例如現(xiàn)實(shí)中的電視與電視頻道。所有客戶端包括發(fā)布者和訂閱者，主題中的消息可以被所有的訂閱者消費(fèi)，消費(fèi)者只能消費(fèi)訂閱之后發(fā)送到主題中的消息。

ActiveMQ 概念

Broker，消息代理，表示消息隊(duì)列服務(wù)器實(shí)體，接受客戶端連接，提供消息通信的核心服務(wù)。
Producer，消息生產(chǎn)者，業(yè)務(wù)的發(fā)起方，負(fù)責(zé)生產(chǎn)消息并傳輸給 Broker 。
Consumer，消息消費(fèi)者，業(yè)務(wù)的處理方，負(fù)責(zé)從 Broker 獲取消息并進(jìn)行業(yè)務(wù)邏輯處理。
Topic，主題，發(fā)布訂閱模式下的消息統(tǒng)一匯集地，不同生產(chǎn)者向 Topic 發(fā)送消息，由 Broker 分發(fā)到不同的訂閱者，實(shí)現(xiàn)消息的廣播。
Queue，隊(duì)列，點(diǎn)對點(diǎn)模式下特定生產(chǎn)者向特定隊(duì)列發(fā)送消息，消費(fèi)者訂閱特定隊(duì)列接收消息并進(jìn)行業(yè)務(wù)邏輯處理。
Message，消息體，根據(jù)不同通信協(xié)議定義的固定格式進(jìn)行編碼的數(shù)據(jù)包，來封裝業(yè)務(wù) 數(shù)據(jù)，實(shí)現(xiàn)消息的傳輸。

ActiveMQ 工程實(shí)例

下面是使用 ActiveMQ 的隊(duì)列模式和發(fā)布-訂閱模式的 Java 代碼示例。

POM 依賴

? ? ? ? ? ? ? ?<dependency> ? ? ? ? ? ?<groupId>org.apache.activemqgroupId> ? ? ? ? ? ?<artifactId>activemq-allartifactId> ? ? ? ? ? ?<version>5.15.5version> ? ? ? ?dependency>

隊(duì)列模式消費(fèi)者

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

/**
?*?

?*?消息消費(fèi)者，用于消費(fèi)消息
?*
?*?@Author?niujinpeng
?*?@Date?2018/9/4?23:45
?*/

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.創(chuàng)建ConnectionFactory
????????ActiveMQConnectionFactory?connectionFactory?=?new?ActiveMQConnectionFactory();
????????//?2.創(chuàng)建Connection
????????Connection?connection?=?connectionFactory.createConnection();
????????//?3.啟動(dòng)連接
????????connection.start();

????????//?4.創(chuàng)建會(huì)話,false，不使用事務(wù)，自動(dòng)應(yīng)答模式
????????Session?session?=?connection.createSession(false,?Session.AUTO_ACKNOWLEDGE);
????????//?5.創(chuàng)建一個(gè)目標(biāo)
????????Destination?destination?=?session.createQueue(queueName);
????????//?6.創(chuàng)建消費(fèi)者
????????MessageConsumer?consumer?=?session.createConsumer(destination);

????????//?7.創(chuàng)建一個(gè)監(jiān)聽器
????????consumer.setMessageListener(new?MessageListener()?{
????????????public?void?onMessage(Message?message)?{
????????????????TextMessage?textMessage?=?(TextMessage)?message;
????????????????try?{
????????????????????System.out.println("接收消息："?+?textMessage.getText());
????????????????}?catch?(JMSException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????}
????????});

????????//?8.關(guān)閉連接
????????//connection.close();
????}
}

隊(duì)列模式生產(chǎn)者

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

/**
?*?

?*?消息提供者，用于向消息中間件發(fā)送消息
?*
?*?@Author?niujinpeng
?*?@Date?2018/9/4?23:28
?*/

public?class?AppProducer?{

????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.創(chuàng)建ConnectionFactory
????????ActiveMQConnectionFactory?connectionFactory?=?new?ActiveMQConnectionFactory();
????????//?2.創(chuàng)建Connection
????????Connection?connection?=?connectionFactory.createConnection();
????????//?3.啟動(dòng)連接
????????connection.start();

????????//?4.創(chuàng)建會(huì)話,false，不使用事務(wù)，自動(dòng)應(yīng)答模式
????????Session?session?=?connection.createSession(false,?Session.AUTO_ACKNOWLEDGE);
????????//?5.創(chuàng)建一個(gè)目標(biāo)
????????Destination?destination?=?session.createQueue(queueName);
????????//?6.創(chuàng)建生產(chǎn)者
????????MessageProducer?producer?=?session.createProducer(destination);

????????//?7.創(chuàng)建消息并發(fā)送
????????for?(int?i?=?0;?i?10;?i++)?{
????????????//?創(chuàng)建消息
????????????TextMessage?textMessage?=?session.createTextMessage("textMessage"?+?i);
????????????//?發(fā)布消息
????????????producer.send(textMessage);
????????????System.out.println("發(fā)送消息："?+?textMessage.getText());
????????}

????????//?8.關(guān)閉連接
????????connection.close();

????}
}

隊(duì)列模式生產(chǎn)者

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

/**
?*?

?*?消息提供者，用于向消息中間件發(fā)送消息
?*
?*?@Author?niujinpeng
?*?@Date?2018/9/4?23:28
?*/

發(fā)布訂閱模式生產(chǎn)者

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

/**
?*?

?*?主題模式
?*?消息消費(fèi)者，用于消費(fèi)消息
?*
?*?@Author?niujinpeng
?*?@Date?2018/9/4?23:45
?*/

public?class?AppConsumer?{

????private?static?final?String?url?=?"tcp://127.0.0.1:61616";
????private?static?final?String?topicName?=?"topic-test";

????public?static?void?main(String[]?args)?throws?JMSException?{
????????//?1.創(chuàng)建ConnectionFactory
????????ActiveMQConnectionFactory?connectionFactory?=?new?ActiveMQConnectionFactory();
????????//?2.創(chuàng)建Connection
????????Connection?connection?=?connectionFactory.createConnection();
????????//?3.啟動(dòng)連接
????????connection.start();

????????//?4.創(chuàng)建會(huì)話,false，不使用事務(wù)，自動(dòng)應(yīng)答模式
????????Session?session?=?connection.createSession(false,?Session.AUTO_ACKNOWLEDGE);
????????//?5.創(chuàng)建一個(gè)目標(biāo)
????????Destination?destination?=?session.createTopic(topicName);
????????//?6.創(chuàng)建消費(fèi)者
????????MessageConsumer?consumer?=?session.createConsumer(destination);

????????//?7.創(chuàng)建一個(gè)監(jiān)聽器
????????consumer.setMessageListener(new?MessageListener()?{
????????????public?void?onMessage(Message?message)?{
????????????????TextMessage?textMessage?=?(TextMessage)?message;
????????????????try?{
????????????????????System.out.println("接收消息："?+?textMessage.getText());
????????????????}?catch?(JMSException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????}
????????});

????????//?8.關(guān)閉連接
????????//connection.close();
????}
}

發(fā)布訂閱模式生產(chǎn)者

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

/**
?*?

?*?主題模式
?*?消息提供者，用于向消息中間件發(fā)送消息
?*
?*?@Author?niujinpeng
?*?@Date?2018/9/4?23:28
?*/

public?class?AppProducer?{

????private?static?final?String?url?=?"tcp://127.0.0.1:61616";
????private?static?final?String?topicName?=?"topic-test";

????public?static?void?main(String[]?args)?throws?JMSException?{
????????//?1.創(chuàng)建ConnectionFactory
????????ActiveMQConnectionFactory?connectionFactory?=?new?ActiveMQConnectionFactory();
????????//?2.創(chuàng)建Connection
????????Connection?connection?=?connectionFactory.createConnection();
????????//?3.啟動(dòng)連接
????????connection.start();

????????//?4.創(chuàng)建會(huì)話,false，不使用事務(wù)，自動(dòng)應(yīng)答模式
????????Session?session?=?connection.createSession(false,?Session.AUTO_ACKNOWLEDGE);
????????//?5.創(chuàng)建一個(gè)目標(biāo)
????????Destination?destination?=?session.createTopic(topicName);
????????//?6.創(chuàng)建生產(chǎn)者
????????MessageProducer?producer?=?session.createProducer(destination);

????????//?7.創(chuàng)建消息并發(fā)送
????????for?(int?i?=?0;?i?10;?i++)?{
????????????//?創(chuàng)建消息
????????????TextMessage?textMessage?=?session.createTextMessage("textMessage"?+?i);
????????????//?發(fā)布消息
????????????producer.send(textMessage);
????????????System.out.println("發(fā)送消息："?+?textMessage.getText());
????????}

????????//?8.關(guān)閉連接
????????connection.close();

????}
}

GitHub源碼：

https://github.com/niumoo/message-queue

Spring 整合 ActiveMQ

在 Spring 中配置 Active MQ 就像Spring 整合其他功能一樣，我們需要在 XML 配置中配置幾個(gè)關(guān)鍵的實(shí)例即可。在 Active MQ 中有幾個(gè)對象的實(shí)例是至關(guān)重要的，如 Active MQ jms 連接工廠，為了減少連接斷開性能時(shí)間消耗的 jms 連接池以及生產(chǎn)者消費(fèi)者等。

下面是一些詳細(xì)說明。

ConnectionFactory 用于管理連接的連接工廠(Spring提供)。
一個(gè) Spring 為我們提供的連接池。
JmsTemplate 每次發(fā)送都會(huì)重新創(chuàng)建連接，會(huì)話和 Productor。
Spring 中提供了SingleConnectionFactory 和CachingConnectionFactory(增加了緩存功能)。
JmsTemplate 是用于發(fā)送和接收消息的模板類。
是spring提供的，只需要向Spring 容器內(nèi)注冊這個(gè)類就可以使用 JmsTemplate 方便的操作jms。
JmsTemplate 類是線程安全的，可以在整個(gè)應(yīng)用范圍使用。
MessageListerner 消息監(jiān)聽器
使用一個(gè)onMessage方法，該方法只接收一個(gè)Message參數(shù)。

POM 依賴

<properties>
????????<spring.version>5.0.4.RELEASEspring.version>
????properties>
????<dependencies>

????????<dependency>
????????????<groupId>junitgroupId>
????????????<artifactId>junitartifactId>
????????????<version>4.11version>
????????????<scope>testscope>
????????dependency>

????????<dependency>
????????????<groupId>org.springframeworkgroupId>
????????????<artifactId>spring-contextartifactId>
????????????<version>${spring.version}version>
????????dependency>

????????<dependency>
????????????<groupId>org.springframeworkgroupId>
????????????<artifactId>spring-jmsartifactId>
????????????<version>5.1.1.RELEASEversion>
????????dependency>

????????<dependency>
????????????<groupId>org.springframeworkgroupId>
????????????<artifactId>spring-testartifactId>
????????????<version>${spring.version}version>
????????dependency>

????????
????????<dependency>
????????????<groupId>javax.jmsgroupId>
????????????<artifactId>javax.jms-apiartifactId>
????????????<version>2.0.1version>
????????dependency>

????????<dependency>
????????????<groupId>org.apache.activemqgroupId>
????????????<artifactId>activemq-coreartifactId>
????????????<version>5.7.0version>
????????????<exclusions>
????????????????<exclusion>
????????????????????<artifactId>spring-contextartifactId>
????????????????????<groupId>org.springframeworkgroupId>
????????????????exclusion>
????????????????<exclusion>
????????????????????<groupId>org.apache.geronimo.specsgroupId>
????????????????????<artifactId>geronimo-jms_1.1_specartifactId>
????????????????exclusion>
????????????exclusions>
????????dependency>
????dependencies>

XML 配置

XML 公共配置

為了份文件配置方便管理，下面是提取出來的公共配置，為了在獨(dú)立配置生產(chǎn)者和消費(fèi)者 XML文件時(shí)引入，當(dāng)然也可以直接把生產(chǎn)者和消費(fèi)者以及所有的 XML bean 配置在一個(gè)文件里。

<?xml ?version="1.0"?encoding="UTF-8"?>
<beans?xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:content="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans?http://www.springframework.org/schema/beans/spring-beans.xsd?http://www.springframework.org/schema/context?http://www.springframework.org/schema/context/spring-context.xsd">

????<content:annotation-config/>

????
????<bean?id="targerConnectionFactory"?class="org.apache.activemq.ActiveMQConnectionFactory">
????????<property?name="brokerURL"?value="tcp://127.0.0.1:61616"/>
????bean>

????
????<bean?id="connectionFactory"?class="org.springframework.jms.connection.SingleConnectionFactory">
????????<property?name="targetConnectionFactory"?ref="targerConnectionFactory"/>
????bean>

????
????<bean?id="queueDestination"?class="org.apache.activemq.command.ActiveMQQueue">
????????<constructor-arg?value="queue-spring"/>
????bean>

????
????<bean?id="topicDestination"?class="org.apache.activemq.command.ActiveMQTopic">
????????<constructor-arg?value="topic-spring"/>
????bean>

beans>

XML 消費(fèi)者

消費(fèi)者主要是一個(gè)消息監(jiān)聽器，監(jiān)聽指定的隊(duì)列或者主題的消息信息，來有消息時(shí)調(diào)用回調(diào)監(jiān)聽處理方法。這里我注釋掉了監(jiān)聽的隊(duì)列模式，指定了主題模式。

<?xml ?version="1.0"?encoding="UTF-8"?>
<beans?xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:content="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans?http://www.springframework.org/schema/beans/spring-beans.xsd?http://www.springframework.org/schema/context?http://www.springframework.org/schema/context/spring-context.xsd">

????
????<import?resource="common.xml"/>

????
????<bean?id="consumerMessageListener"?class="net.codingme.jms.consumer.ConsumerMessageListener"/>

????
????<bean?id="jmsContainer"?class="org.springframework.jms.listener.DefaultMessageListenerContainer">
????????<property?name="connectionFactory"?ref="connectionFactory"/>
????????
????????
????????
????????<property?name="destination"?ref="topicDestination"/>
????????<property?name="messageListener"?ref="consumerMessageListener"/>
????bean>

beans>

XML 生產(chǎn)者

生成者的配置主要是使用 spring jms 模版對象，創(chuàng)建生產(chǎn)者實(shí)例用于生產(chǎn)消息。

生產(chǎn)者編寫

1. 定義接口

package?net.codingme.jms.producer;

/**
?*?

?*
?*?@Author?niujinpeng
?*?@Date?2018/11/2518:19
?*/

public?interface?ProducerService?{
????public?void?sendMessage(String?message);
}

2. 主題模式生產(chǎn)者

package?net.codingme.jms.producer;

import?org.springframework.beans.factory.annotation.Autowired;
import?org.springframework.jms.core.JmsTemplate;
import?org.springframework.jms.core.MessageCreator;

import?javax.annotation.Resource;
import?javax.jms.*;

/**
?*?

?*
?*?@Author?niujinpeng
?*?@Date?2018/11/25?19:24
?*/

public?class?ProducerServiceImpl?implements?ProducerService?{

????@Autowired
????JmsTemplate?jmsTemplate;
????/**
?????*?主題模式
?????*/
????@Resource(name?=?"topicDestination")
????Destination?destination;

????@Override
????public?void?sendMessage(String?message)?{
????????//?使用jmsTemplate發(fā)送消息
????????jmsTemplate.send(destination,?new?MessageCreator()?{
????????????//?創(chuàng)建消息
????????????@Override
????????????public?Message?createMessage(Session?session)?throws?JMSException?{
????????????????TextMessage?textMessage?=?session.createTextMessage(message);
????????????????return?textMessage;
????????????}
????????});
????????System.out.println("發(fā)送消息："?+?message);

????}
}

3. Spring 啟動(dòng) 生產(chǎn)者

package?net.codingme.jms.producer;

import?org.springframework.context.support.ClassPathXmlApplicationContext;

/**
?*?

?*?啟動(dòng)器
?*
?*?@Author?niujinpeng
?*?@Date?2018/11/25?21:48
?*/

public?class?AppProducer?{

????public?static?void?main(String[]?args)?{
????????//?裝載配置文件
????????ClassPathXmlApplicationContext?context?=?new?ClassPathXmlApplicationContext("classpath:producer.xml");
????????ProducerService?service?=?context.getBean(ProducerService.class);

????????for?(int?i?=?0;?i?10;?i++)?{
????????????service.sendMessage("test"?+?i);
????????}
????????context.close();
????}

}

消費(fèi)者編寫

Spring啟動(dòng)和生產(chǎn)者類似。下面是消費(fèi)者監(jiān)聽器的實(shí)現(xiàn)。

package?net.codingme.jms.consumer;

import?javax.jms.JMSException;
import?javax.jms.Message;
import?javax.jms.MessageListener;
import?javax.jms.TextMessage;

/**
?*?

?*?消息監(jiān)聽器
?*
?*?@Author?niujinpeng
?*?@Date?2018/11/25?22:28
?*/

public?class?ConsumerMessageListener?implements?MessageListener?{
????@Override
????public?void?onMessage(Message?message)?{
????????TextMessage?textMessage?=?(TextMessage)?message;
????????try?{
????????????System.out.println("接收消息："?+?textMessage.getText());
????????}?catch?(JMSException?e)?{
????????????e.printStackTrace();
????????}
????}
}

運(yùn)行測試

首先主題模式下啟動(dòng)兩個(gè)消費(fèi)者，使用生產(chǎn)者推送10條消息。

在每個(gè)消費(fèi)者下面都可以看到推送的完整消息。

文中代碼已經(jīng)上傳到GitHub：

https://github.com/niumoo/message-queue

推薦閱讀

(點(diǎn)擊標(biāo)題可跳轉(zhuǎn)閱讀)

夯實(shí)Java基礎(chǔ)系列16：一文讀懂Java IO流和常見面試題

夯實(shí)Java基礎(chǔ)系列15：Java注解簡介和最佳實(shí)踐

夯實(shí)Java基礎(chǔ)系列14：深入理解Java枚舉類

夯實(shí)Java基礎(chǔ)系列11：深入理解Java中的回調(diào)機(jī)制

夯實(shí)Java基礎(chǔ)系列10：深入理解Java中的異常體系

夯實(shí)Java基礎(chǔ)系列9：深入理解Class類和Object類

夯實(shí)Java基礎(chǔ)系列8：深入理解Java內(nèi)部類及其實(shí)現(xiàn)原理

夯實(shí)Java基礎(chǔ)系列7：一文讀懂Java 代碼塊和代碼執(zhí)行順序

一文搞懂抽象類和接口，從基礎(chǔ)到面試題，揭秘其本質(zhì)區(qū)別！

一文讀懂 Java 文件和包結(jié)構(gòu)，解讀開發(fā)中常用的 jar 包

一文了解 final 關(guān)鍵字的特性、使用方法以及實(shí)現(xiàn)原理

點(diǎn)個(gè)“在看”，轉(zhuǎn)發(fā)朋友圈，都是對我最好的支持！

總結(jié)

以上是生活随笔為你收集整理的arouter跨module传递消息_消息队列中间件（二）使用 ActiveMQ的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： window服务器cpu过高的排查_高频
下一篇：配置库用户_手把手教新手安装Anacon