當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

activemq安装与配置_ActiveMQ(消息队列)从入门到了解

發布時間：2023/12/1 编程问答 40 豆豆

生活随笔收集整理的這篇文章主要介紹了 activemq安装与配置_ActiveMQ(消息队列)从入门到了解小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

一、 ActiveMQ 簡介

1 什么是 ActiveMQ

ActiveMQ 是 Apache 出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持 JMS1.1 和 J2EE 1.4 規范的 JMS Provider 實現，盡管 JMS 規范出臺已經是很久的事情了，但是 JMS 在當今的 J2EE 應用中間仍然扮演著特殊的地位。

2 什么是消息

“消息”是在兩臺計算機間傳送的數據單位。消息可以非常簡單，例如只包含文本字符串；也可以更復雜，可能包含嵌入對象。

3 什么是隊列

4 什么是消息隊列

“消息隊列”是在消息的傳輸過程中保存消息的容器。

5 常用消息服務應用

5.1 ActiveMQ

ActiveMQ 是 Apache 出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持 JMS1.1 和 J2EE 1.4 規范的 JMS Provider 實現。

5.2 RabbitMQ

RabbitMQ 是一個在 AMQP 基礎上完成的，可復用的企業消息系統。他遵循 Mozilla Public License 開源協議。開發語言為 Erlang。

5.3 RocketMQ

由阿里巴巴定義開發的一套消息隊列應用服務。

四、 ActiveMQ 安裝

1 下載資源

ActiveMQ 官網： ActiveMQ

1.1 版本說明

ActiveMQ5.10.x 以上版本必須使用 JDK1.8 才能正常使用。 ActiveMQ5.9.x 及以下版本使用 JDK1.7 即可正常使用。

2 上傳至 Linux 服務器

3 解壓安裝文件

tar -zxf apache-activemq-5.9.0-bin.tar.gz

4 檢查權限

ls -al apache-activemq-5.9.0/bin 如果權限不足,則無法執行,需要修改文件權限: chmod 755 activemq

5 復制應用至本地目錄

cp -r apache-activemq-5.9.0 /usr/local/activemq

6 啟動 ActiveMQ

/usr/local/activemq/bin/activemq start

7 測試 ActiveMQ

7.1 檢查進程

ps aux | grep activemq 見到下述內容即代表啟動成功

7.2 管理界面

使用瀏覽器訪問 ActiveMQ 管理應用, 地址如下: http://ip:8161/admin/

用戶名: admin 密碼: admin

ActiveMQ 使用的是 jetty 提供 HTTP 服務.啟動稍慢,建議短暫等待再訪問測試. 見到如下界面代表服務啟動成功

7.3 修改訪問端口

修改 ActiveMQ 配置文件: /usr/local/activemq/conf/jetty.xml

配置文件修改完畢，保存并重新啟動 ActiveMQ 服務。 7.4 修改用戶名和密碼

修改 conf/users.properties 配置文件.內容為: 用戶名=密碼保存并重啟 ActiveMQ 服務即可.

8 重啟 ActiveMQ

/usr/local/activemq/bin/activemq restart

9 關閉 ActiveMQ

/usr/local/activemq/bin/activemq stop

10 配置文件 activemq.xml

配置文件中,配置的是 ActiveMQ 的核心配置信息. 是提供服務時使用的配置. 可以修改啟動的訪問端口. 即 java 編程中訪問 ActiveMQ 的訪問端口. 默認端口為 61616. 使用協議是: tcp 協議. 修改端口后, 保存并重啟 ActiveMQ 服務即可.

11 ActiveMQ 目錄介紹

從它的目錄來說,還是很簡單的: * bin 存放的是腳本文件 * conf 存放的是基本配置文件 * data 存放的是日志文件 * docs 存放的是說明文檔 * examples 存放的是簡單的實例 * lib 存放的是 activemq 所需 jar 包 * webapps 用于存放項目的目錄

五、 ActiveMQ 術語

1 Destination

目的地，JMS Provider（消息中間件）負責維護，用于對 Message 進行管理的對象。 MessageProducer 需要指定 Destination 才能發送消息，MessageReceiver 需要指定 Destination 才能接收消息。

2 Producer

消息生成者，負責發送 Message 到目的地。

3 Consumer | Receiver

消息消費者，負責從目的地中消費【處理|監聽|訂閱】Message。

4 Message

消息，消息封裝一次通信的內容。

六、 ActiveMQ 應用

1 ActiveMQ 常用 API 簡介

下述 API 都是接口類型,由定義在 javax.jms 包中. 是 JMS 標準接口定義.

1.1 ConnectionFactory

鏈接工廠, 用于創建鏈接的工廠類型.

1.2 Connection

鏈接. 用于建立訪問 ActiveMQ 連接的類型, 由鏈接工廠創建.

1.3 Session

會話, 一次持久有效有狀態的訪問. 由鏈接創建.

1.4 Destination & Queue

目的地, 用于描述本次訪問 ActiveMQ 的消息訪問目的地. 即 ActiveMQ 服務中的具體隊列. 由會話創建. interface Queue extends Destination

1.5 MessageProducer

消息生成者, 在一次有效會話中, 用于發送消息給 ActiveMQ 服務的工具. 由會話創建.

1.6 MessageConsumer

消息消費者【消息訂閱者，消息處理者】, 在一次有效會話中, 用于從 ActiveMQ 服務中獲取消息的工具. 由會話創建.

1.7 Message

消息, 通過消息生成者向 ActiveMQ 服務發送消息時使用的數據載體對象或消息消費者從 ActiveMQ 服務中獲取消息時使用的數據載體對象. 是所有消息【文本消息，對象消息等】具體類型的頂級接口. 可以通過會話創建或通過會話從 ActiveMQ 服務中獲取.

2 JMS-HelloWorld

2.1 處理文本消息

2.1.1 創建消息生產者

2.1.1.2 修改 POM 文件添加 ActiveMQ 坐標

2.1.1.3 編寫消息的生產者

package com.sxt;import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session;import org.apache.activemq.ActiveMQConnectionFactory;public class HelloWorldProducer {/*** 生產消息* @param MsgTest*/public void sendHelloWorldActiveMQ(String MsgTest){//定義鏈接工廠ConnectionFactory connectionFactory = null;//定義鏈接對象Connection connection = null;//定義會話Session session = null;//定義目的地Destination destination = null;//定義消息的發送者MessageProducer producer = null;//定義消息Message message = null;try {/*** userName:訪問ActiveMQ服務的用戶名。用戶密碼。默認的為admin。用戶名可以通過jetty-ream.properties文件進行修改* password:訪問ActiveMQ服務的用戶名。用戶密碼。默認的為admin。用戶名可以通過jetty-ream.properties文件進行修改* brokerURL:訪問ActiveMQ服務的路徑地址。路徑結構為:協議名://主機地址:端口號*/connectionFactory = new ActiveMQConnectionFactory("admin","admin", "tcp://192.168.44.130:61616");//創建連接對象connection = connectionFactory.createConnection();//啟動連接connection.start();/*** transacted:是否使用事務可選值為：true|false* true:使用事務當設置次變量值。Session.SESSION_TRANSACTED* false:不適用事務,設置次變量則acknowledgeMode參數必須設置* acknowledgeMode:* Session.AUTO_ACKNOWLEDGE:自動消息確認機制* Session.CLIENT_ACKNOWLEDGE:客戶端確認機制* Session.DUPS_OK_ACKNOWLEDGE:有副本的客戶端確認消息機制*/session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//創建目的地，目的地即隊列的名稱。消息的消費者需要通過該名稱訪問對應的隊列destination = session.createQueue("Helloworld-destination");//創建消息的生產者producer = session.createProducer(destination);//創建消費者對象message = session.createTextMessage(MsgTest);//發送消息producer.send(message);} catch (JMSException e) {e.printStackTrace();}finally{if(producer != null){try {producer.close();} catch (JMSException e) {e.printStackTrace();}}if(session != null){try {session.close();} catch (JMSException e) {e.printStackTrace();}}if(connection != null ){try {connection.close();} catch (JMSException e) {e.printStackTrace();}}}}}

2.1.2 創建消息消費者

2.1.2.1 創建工程

2.1.2.2 修改 POM 文件添加 ActiveMQ 坐標

2.1.2.3 編寫消息的消費者

package com.sxt;import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;public class HelloWorldConsumer {/*** 消費消息*/public void readHelloWorldActiveMQ(){//定義鏈接工廠ConnectionFactory connectionFactory = null;//定義鏈接對象Connection connection = null;//定義會話Session session = null;//目的地Destination destination = null;//定義消息的消費者MessageConsumer consumer = null;//定義消息Message message = null;try {/*** userName:訪問ActiveMQ服務的用戶名。用戶密碼。默認的為admin。用戶名可以通過jetty-ream.properties文件進行修改* password:訪問ActiveMQ服務的用戶名。用戶密碼。默認的為admin。用戶名可以通過jetty-ream.properties文件進行修改* brokerURL:訪問ActiveMQ服務的路徑地址。路徑結構為:協議名://主機地址:端口號*/connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.44.130:61616");//創建連接對象connection = connectionFactory.createConnection();//啟動連接connection.start();/*** transacted:是否使用事務可選值為：true|false* true:使用事務當設置次變量值。Session.SESSION_TRANSACTED* false:不適用事務,設置次變量則acknowledgeMode參數必須設置* acknowledgeMode:* Session.AUTO_ACKNOWLEDGE:自動消息確認機制* Session.CLIENT_ACKNOWLEDGE:客戶端確認機制* Session.DUPS_OK_ACKNOWLEDGE:有副本的客戶端確認消息機制*/session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//創建目的地，目的地名稱即隊列的名稱。消息的消費者需要通過此名稱訪問對應的隊列destination = session.createQueue("Helloworld-destination");//創建消息的消費者consumer = session.createConsumer(destination);//接收消息message = consumer.receive();//處理消息String msg = ((TextMessage)message).getText();System.out.println("從ActiveMQ服務中獲得的消息"+msg);} catch (JMSException e) {e.printStackTrace();}finally{// 回收消息發送者資源if(consumer != null){try {consumer.close();} catch (JMSException e) {e.printStackTrace();}}if(session != null){try {session.close();} catch (JMSException e) {e.printStackTrace();}} if(connection != null ){try {connection.close();} catch (JMSException e) {e.printStackTrace();}} }} }

2.1.3 測試

2.1.3.1 Producer

package com.sxt;public class Test {public static void main(String[] args) {System.out.println("消息發送方..."); // HelloWorldProducer hwp = new HelloWorldProducer(); // hwp.sendHelloWorldActiveMQ("HelloWorld!");/* Users user = new Users();user.setUserid(3); user.setUserName("多對的");user.setPassWord("46464");HelloWorldProducer2 producer = new HelloWorldProducer2();producer.sendHelloWorldActiveMQ(user);}*/HelloWorldProducer3 producer = new HelloWorldProducer3();producer.sendHelloWorldActiveMQ("Helloworlddsdfsdfdsf");} }

2.1.3.2 Consumer

package com.sxt;public class Test {public static void main(String[] args) {System.out.println("消息接收方..."); // HelloWorldConsumer hc = new HelloWorldConsumer(); // hc.readHelloWorldActiveMQ(); // HelloWorldConsumer2 consumer = new HelloWorldConsumer2(); // consumer.readHelloWorldActiveMQ();HelloWorldConsumer3 consumer = new HelloWorldConsumer3();consumer.readHelloWorldActiveMQ();} }

2.2 處理對象消息

2.2.1 定義消息對象

package com.sxt;import java.io.Serializable;public class Users implements Serializable{private static final long serialVersionUID = -1305446877847351356L;private int userid;private String userName;private String passWord;public int getUserid() {return userid;}public void setUserid(int userid) {this.userid = userid;}public String getUserName() {return userName;}public void setUserName(String userName) {this.userName = userName;}public String getPassWord() {return passWord;}public void setPassWord(String passWord) {this.passWord = passWord;} }

2.2.2 創建生產者

大同小異，需要在上面的基礎上修改

2.2.3 定義消息消費者

3 JMS - 實現隊列服務監聽

隊列服務監聽使用的觀察者設計模式

3.1 創建消息生產者

僅只修改隊列名稱

3.2 消息消費者

并取消資源關閉

保持開啟狀態

4 Topic 模型

4.1 Publish/Subscribe 處理模式（Topic）

消息生產者（發布）將消息發布到 topic 中，同時有多個消息消費者（訂閱）消費該消息。和點對點方式不同，發布到 topic 的消息會被所有訂閱者消費。當生產者發布消息，不管是否有消費者。都不會保存消息一定要先有消息的消費者，后有消息的生產者。

4.2 創建生產者

原來的生產者為：

4.3 創建消費者

package com.bjsxt;import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;public class HelloWorldConsumerTopic1 implements Runnable{/*** 消費消息*/public void readHelloWorldActiveMQ() {// 定義鏈接工廠ConnectionFactory connectionFactory = null;// 定義鏈接對象Connection connection = null;// 定義會話Session session = null;// 目的地Destination destination = null;// 定義消息的發送者MessageConsumer consumer = null;// 定義消息Message message = null;try {/*** userName:訪問ActiveMQ服務的用戶名。用戶密碼。默認的為admin。用戶名可以通過jetty-ream.* properties文件進行修改* password:訪問ActiveMQ服務的用戶名。用戶密碼。默認的為admin。用戶名可以通過jetty-ream.* properties文件進行修改 brokerURL:訪問ActiveMQ服務的路徑地址。路徑結構為:協議名://主機地址:端口號*/connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.70.151:61616");// 創建連接對象connection = connectionFactory.createConnection();// 啟動連接connection.start();/*** transacted:是否使用事務可選值為：true|false true:使用事務* 當設置次變量值。Session.SESSION_TRANSACTED false:不適用事務,設置次變量* 則acknowledgeMode參數必須設置 acknowledgeMode:* Session.AUTO_ACKNOWLEDGE:自動消息確認機制* Session.CLIENT_ACKNOWLEDGE:客戶端確認機制* Session.DUPS_OK_ACKNOWLEDGE:有副本的客戶端確認消息機制*/session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 創建目的地，目的地名稱即隊列的名稱。消息的消費者需要通過此名稱訪問對應的隊列destination = session.createTopic("test-topic");// 創建消息的消費者consumer = session.createConsumer(destination);consumer.setMessageListener(new MessageListener() {//ActiveMQ 回調的方法。通過該方法將消息傳遞到 consumer@Overridepublic void onMessage(Message message) {//處理消息String msg=null;try {msg = ((TextMessage)message).getText();} catch (JMSException e) {// TODO Auto-generated catch blocke.printStackTrace();}System.out.println("從ActiveMQ服務中獲取的文本信息 ---topic1 "+msg);}}); } catch (Exception e) {e.printStackTrace();} }@Overridepublic void run() {this.readHelloWorldActiveMQ();} }

Spring-activemq-consumer

pom.xml:

myMessageListener:

package com.sxt.listener;import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.ObjectMessage;import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;import com.sxt.pojo.Users; import com.sxt.service.UserService;/*** 消息服務監聽器* @author Administrator**/ @Component(value="myListener") public class MyMessageListener implements MessageListener{@Autowiredprivate UserService userService;@Overridepublic void onMessage(Message message) {//處理消息ObjectMessage objectMessage = (ObjectMessage) message;Users user = null;try {user = (Users)objectMessage.getObject();} catch (JMSException e) {e.printStackTrace();}this.userService.showUser(user);}}

applicationContext-jms.xml

<?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:jms="http://www.springframework.org/schema/jms" xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/jmshttp://www.springframework.org/schema/jms/spring-jms.xsdhttp://activemq.apache.org/schema/corehttp://activemq.apache.org/schema/core/activemq-core.xsd"><amq:connectionFactory brokerURL="tcp://192.168.44.130:61616"userName="admin" password="admin" id="amqConnectionFactory"/><bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"><property name="targetConnectionFactory" ref="amqConnectionFactory"></property><property name="sessionCacheSize" value="3"></property></bean><jms:listener-container acknowledge="auto" container-type="default"destination-type="queue" connection-factory="connectionFactory" ><jms:listener destination="test-spring" ref="myListener"/></jms:listener-container></beans>

spring-activemq-producer(生產者)

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>com.sxt</groupId><artifactId>parent</artifactId><version>0.0.1-SNAPSHOT</version></parent><groupId>com.sxt</groupId><artifactId>spring-activemq-producer</artifactId><version>0.0.1-SNAPSHOT</version><packaging>war</packaging><dependencies><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId></dependency><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId></dependency><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId></dependency><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-jms-pool</artifactId></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-beans</artifactId></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-webmvc</artifactId></dependency><dependency><groupId>jstl</groupId><artifactId>jstl</artifactId></dependency><dependency><groupId>javax.servlet</groupId><artifactId>servlet-api</artifactId><scope>provided</scope></dependency><dependency><groupId>javax.servlet</groupId><artifactId>jsp-api</artifactId><scope>provided</scope></dependency></dependencies><build><plugins><plugin><groupId>org.apache.tomcat.maven</groupId><artifactId>tomcat7-maven-plugin</artifactId><configuration><path>/</path><port>8080</port></configuration></plugin></plugins></build> </project>

userServiceImpl:

package com.sxt.service.impl;import javax.jms.JMSException; import javax.jms.Message; import javax.jms.ObjectMessage; import javax.jms.Session;import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import org.springframework.stereotype.Service;import com.sxt.pojo.Users; import com.sxt.service.UserService;@Service public class UserServiceImpl implements UserService{@Autowiredprivate JmsTemplate jmsTemplate;@Overridepublic void addUser(final Users user) {//發送消息this.jmsTemplate.send(new MessageCreator() {@Overridepublic Message createMessage(Session session) throws JMSException {ObjectMessage message = session.createObjectMessage(user);return message;}});} }

applicationContext-jms.xml

<?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:jms="http://www.springframework.org/schema/jms"xmlns:context="http://www.springframework.org/schema/context"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/jmshttp://www.springframework.org/schema/jms/spring-jms.xsdhttp://activemq.apache.org/schema/corehttp://activemq.apache.org/schema/core/activemq-core.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context.xsd"><amq:connectionFactory brokerURL="tcp://192.168.44.130:61616"userName="admin" password="admin" id="amqConnectionFactory" /><bean id="pooledConnectionFactoryBean" class="org.apache.activemq.pool.PooledConnectionFactoryBean"><property name="connectionFactory" ref="amqConnectionFactory"></property><property name="maxConnections" value="10"></property></bean><bean id="connectionFactory"class="org.springframework.jms.connection.CachingConnectionFactory"><property name="targetConnectionFactory" ref="amqConnectionFactory"></property><property name="sessionCacheSize" value="3"></property></bean><bean id="template" class="org.springframework.jms.core.JmsTemplate"><property name="connectionFactory" ref="connectionFactory"></property><property name="defaultDestinationName" value="test-spring"></property></bean> </beans>

總結

以上是生活随笔為你收集整理的activemq安装与配置_ActiveMQ(消息队列)从入门到了解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：我的世界黑曜石怎么挖
下一篇： wps单机无网络版_单平台销量破百万，这