當前位置：首頁 > 前端技术 > javascript >内容正文

javascript

RabbitMQ安装和使用（和Spring集成）

發布時間：2025/3/8 javascript 20 豆豆

生活随笔收集整理的這篇文章主要介紹了 RabbitMQ安装和使用（和Spring集成）小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

一、安裝Rabbit MQ　　

　　Rabbit MQ 是建立在強大的Erlang OTP平臺上，因此安裝Rabbit MQ的前提是安裝Erlang。通過下面兩個連接下載安裝3.2.3 版本：

下載并安裝?Eralng OTP For Windows?(vR16B03)

運行安裝?Rabbit MQ Server Windows Installer?(v3.2.3)

具體操作步驟參考：在 Windows 上安裝Rabbit MQ 指南

本人遇到的問題

當安裝RabbitMQ后，使用rabbitmqctl，rabbitmq-plugins等相應的命令，出現如下錯誤：* connected to epmd (port 4369) on mqmaster* epmd reports node 'rabbit' running on port 25672* TCP connection succeeded but Erlang distribution failed* suggestion: hostname mismatch?* suggestion: is the cookie set correctly?* suggestion: is the Erlang distribution using TLS?

　　原因是，安裝 Rabbit MQ的過程中沒有使用默認的安裝路徑。重新安裝并使用默認的安裝路徑就沒有問題了。如果使用自定義路徑，需要同步?Erlang Cookies。具體解釋請參考官網中的解釋：http://www.rabbitmq.com/install-windows-manual.html?（在Finding Command Scripts and App Data 標題下）。

二、Spring集成Rabbit MQ示例1

　　1.創建一個新的maven工程demo1。

　　2.pom文件依賴如下。

<dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>1.6.1.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-beans</artifactId><version>4.3.1.RELEASE</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>4.3.1.RELEASE</version></dependency> View Code

　　3.創建application.properties文件。

#mq mq.username=guest mq.password=guest mq.addresses=127.0.0.1:5672

　　4.創建spring公共配置文件applicationContext.xml。

　　5.創建Rabbit MQ的配置文件。

<?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:context="http://www.springframework.org/schema/context" xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd"><description>rabbitmq 連接服務配置</description><rabbit:connection-factory id="connectionFactory" addresses="${mq.addresses}" username="${mq.username}" password="${mq.password}" /><rabbit:admin connection-factory="connectionFactory"/><rabbit:template id="amqpTemplate" exchange="test-mq-exchange" connection-factory="connectionFactory" message-converter="jsonMessageConverter" /><bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /><rabbit:queue id="test_queue_id" name="test_queue_name" durable="true" auto-delete="false" exclusive="false" /><rabbit:direct-exchange name="test-mq-exchange" durable="true" auto-delete="false" id="test-mq-exchange"><rabbit:bindings><rabbit:binding queue="test_queue_name" key="test_queue_routing_key"/></rabbit:bindings></rabbit:direct-exchange><rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"><rabbit:listener queues="test_queue_name" ref="queueListenter"/></rabbit:listener-container> </beans> View Code

　　6.新建MQ消息實體類（MqMessage.java）。

public class MqMessage implements Serializable {private static final long serialVersionUID = -6791105187137215924L;private String id;private String type;//保留參數private Object params;private Object body;public String getId() {return id;}public void setId(String id) {this.id = id;}public String getType() {return type;}public void setType(String type) {this.type = type;}public Object getBody() {return body;}public void setBody(Object body) {this.body = body;}public Object getParams() {return params;}public void setParams(Object params) {this.params = params;}} View Code

　　7.新建MQ消息發送接口（MQSender.java）。

import com.hjz.rabbit.common.MqMessage; public interface MQSender {/*** 發送消息到指定隊列* @param message* @param queueName*/void sendMessage(MqMessage message, String queueName); }

?　　8.新建MQ消息發送實現類 MQSenderImpl.java。

import java.util.UUID;import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;import com.hjz.rabbit.api.MQSender; import com.hjz.rabbit.common.MqMessage;@Component public class MQSenderImpl implements MQSender {@Autowiredprivate AmqpTemplate amqpTemplate;@Overridepublic void sendMessage(MqMessage message, String queueName) {if(message==null){System.out.println("消息發送失敗：消息為null");return;}if(message.getId()==null){message.setId(UUID.randomUUID().toString()); }amqpTemplate.convertAndSend(queueName, message);} }/*convertAndSend：將Java對象轉換為消息發送到匹配Key的交換機中Exchange，由于配置了JSON轉換，這里是將Java對象轉換成JSON字符串的形式。
　　原文：Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key. */

　　9.創建消息監聽器QueueListenter.java。

import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; import org.springframework.stereotype.Component;@Component("queueListenter") public class QueueListenter implements MessageListener {@Overridepublic void onMessage(Message msg) {try{System.out.print(msg.toString());}catch(Exception e){e.printStackTrace();}}}

　　10.終于到最后一步了，新建測試類。

import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;import com.hjz.rabbit.api.MQSender; import com.hjz.rabbit.common.MqMessage;@RunWith(value = SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = {"classpath*:applicationContext.xml", "classpath*:applicationContext-mq.xml"}) public class TestQueue{@Autowiredprivate MQSender mqSender;final String queueName = "test_queue_routing_key";@Testpublic void send(){MqMessage message = new MqMessage();message.setBody("大家好，我是hjzgg!!!");mqSender.sendMessage(message, queueName);} }

?　　11.最終結果，消息監聽器中的輸出如下。

(Body:'{"id":"723db571-1b0e-4ea5-aad4-dbfed01d158a","type":null,"params":null,"body":"大家好，我是hjzgg!!!"}' MessageProperties [headers={__TypeId__=com.hjz.rabbit.common.MqMessage}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/json, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=test-mq-exchange, receivedRoutingKey=test_queue_routing_key, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-PSySAf57jb5fU99UgoG2kA, consumerQueue=test_queue_name])

?　　12.瀏覽器中輸入“http://localhost:15672/”，進入 Rabbit MQ的管理系統，查看 queue和exchange對應的相關信息如下。

三、Spring集成Rabbit MQ示例2

　　1.對MQ的生產者和消費者做了一些簡單的封裝，同一放到了一個工程下面里。無論是生產者還是消費者，只需要依賴上這個工程對應的jar就可以了。另外就是依賴上這個工程中MQ對應的XML文件。工程的基本結構如下所示。

　　2.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:context="http://www.springframework.org/schema/context" xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd"><rabbit:connection-factory id="myConnectionFactory" addresses="${mq.addresses}" username="${mq.username}" password="${mq.password}" publisher-confirms="false"/><bean id="myJsonMessageConverter" class="com.hjz.mq.converter.FastJsonMessageConverter"></bean><bean id="myRetryConnTemplate" class="org.springframework.retry.support.RetryTemplate"><property name="backOffPolicy"><bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy"><property name="initialInterval" value="500"/><property name="multiplier" value="10.0"/><property name="maxInterval" value="5000"/></bean></property></bean><bean id="myRabbitAdmin" class="org.springframework.amqp.rabbit.core.RabbitAdmin"> <constructor-arg ref="myConnectionFactory" /> </bean> <bean id="myRabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"> <constructor-arg ref="myConnectionFactory"></constructor-arg><property name="messageConverter" ref="myJsonMessageConverter"></property> </bean><bean id="msgSender" class="com.hjz.mq.sender.impl.MsgSenderImpl" /> </beans> View Code

　　3.消費者。

package com.hjz.mq.consumer;import javax.annotation.Resource;import org.apache.commons.lang3.ArrayUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean;import com.alibaba.fastjson.JSON; import com.hjz.mq.common.MqMessage; import com.rabbitmq.client.Channel;public abstract class BaseConsumer implements InitializingBean,ChannelAwareMessageListener,DisposableBean {protected static Logger logger = LoggerFactory.getLogger(BaseConsumer.class); @Resource(name="myRabbitAdmin")private RabbitAdmin rabbitAdmin;@Resource(name="myRabbitTemplate")private RabbitTemplate rabbitTemplate;//@Resource(name="myListenerContainer")private SimpleMessageListenerContainer listenerContainer;private static final String EXCHANGE_NAME = "directExchange";public static final String DEFAULT_CHARSET = "UTF-8";public void afterPropertiesSet(){String[] queueNames = this.getQueueNames();if(ArrayUtils.isEmpty(queueNames)){//throw new RuntimeException("請設置需要消費的隊列");logger.warn("沒有設置需要消費的隊列");return;}// 聲明交換器/*DirectExchange exchange = new DirectExchange(EXCHANGE_NAME);rabbitAdmin.declareExchange(exchange);Binding binding = BindingBuilder.bind(queue).to(exchange).with(getRoutingKey());rabbitAdmin.declareBinding(binding);*///聲明一個監聽容器listenerContainer = new SimpleMessageListenerContainer();listenerContainer.setConnectionFactory(rabbitTemplate.getConnectionFactory());Queue[] queues = new Queue[queueNames.length];//注冊監聽for(int i=0,len=queueNames.length;i<len;i++){String queueName = queueNames[i];Queue queue = new Queue(queueName,true,false,false);rabbitAdmin.declareQueue(queue);queues[i] = queue;}listenerContainer.addQueues(queues);listenerContainer.setPrefetchCount(1);listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);listenerContainer.setMessageListener(this);listenerContainer.start();}public void destroy(){logger.debug("關閉監聽...");if(listenerContainer!=null){listenerContainer.stop();}}@Overridepublic void onMessage(Message message, Channel channel) throws Exception {String msg = new String(message.getBody(), DEFAULT_CHARSET);logger.debug("接收到消息："+msg);MessageProperties msgProps = message.getMessageProperties();//確認接收到消息channel.basicAck(msgProps.getDeliveryTag(), false);try{MqMessage mqMessage = JSON.parseObject(msg, MqMessage.class);if(mqMessage==null || mqMessage.getBody() == null){logger.error("消息體為空，舍棄！");return;}doConsumeMsg(mqMessage);logger.debug("消息消費完成");}catch(Exception ex){logger.error("消息消費失敗:",ex);}}protected abstract void doConsumeMsg(MqMessage mqMessage);protected abstract String[] getQueueNames();// protected abstract String[] getRoutingKey(); } View Code

　　4.生產者。

package com.hjz.mq.sender.impl;import java.util.HashSet; import java.util.Set; import java.util.UUID; import javax.annotation.Resource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate;import com.hjz.mq.common.MqMessage; import com.hjz.mq.sender.MsgSender;public class MsgSenderImpl implements MsgSender {private Logger logger = LoggerFactory.getLogger(MsgSenderImpl.class);@Resource(name="myRabbitAdmin")private RabbitAdmin rabbitAdmin;@Resource(name="myRabbitTemplate")private RabbitTemplate rabbitTemplate;private static final String defaultRouteKey = "";private static final String utf8Encode = "UTF-8";private static Set<String> queueHashSet=new HashSet<String>();/*** appcode：exchange* businessCode：routekey*/@Overridepublic void sendMessage(MqMessage message, String queueName) {if(message==null){logger.warn("消息發送失敗：消息為null");return;}if(!queueHashSet.contains(queueName)){Queue queue = new Queue(queueName,true,false,false);rabbitAdmin.declareQueue(queue);//添加到集合中用于判斷 queueHashSet.add(queueName);}if(message.getId()==null){message.setId(UUID.randomUUID().toString()); }rabbitTemplate.convertAndSend(queueName, message);} } View Code

　　示例中去掉了 Exchange，可以在完整項目中查看關于Exchange的配置。

　　5.消費者web2工程

　　繼承封裝過的MQ消費者抽象類，實現為實現的方法，將消息放入全局application對象中。

package com.hjz.rabbit.consumer;import java.util.ArrayList; import java.util.List; import javax.servlet.ServletContext; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import com.hjz.mq.common.MqConfig; import com.hjz.mq.common.MqMessage; import com.hjz.mq.consumer.BaseConsumer;@Component public class Consumer extends BaseConsumer {@Autowiredprivate ServletContext application;// private ServletContext application = ContextLoader.getCurrentWebApplicationContext().getServletContext(); @SuppressWarnings("unchecked")@Overrideprotected void doConsumeMsg(MqMessage mqMessage) {if(application.getAttribute("MQ_MESSAGE") == null) {application.setAttribute("MQ_MESSAGE", new ArrayList<Object>());}((List<Object>)application.getAttribute("MQ_MESSAGE")).add(mqMessage.getBody());}@Overrideprotected String[] getQueueNames() {return new String[] {MqConfig.MQ_QUEUE_ADD, MqConfig.MQ_QUEUE_SAVE, MqConfig.MQ_QUEUE_UPDATE};} } View Code

　　從全局application對象中獲取消息并返回。

package com.hjz.rabbit.web;import java.util.ArrayList; import java.util.List;import javax.servlet.ServletContext;import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody;import com.alibaba.fastjson.JSONObject; import com.hjz.mq.util.CloneUtils; import com.hjz.mq.util.ReturnCode;@Controller @RequestMapping("consumer") public class MQConsumer {@Autowiredprivate ServletContext application;@SuppressWarnings("unchecked")@RequestMapping("process")@ResponseBodypublic JSONObject consumer() {JSONObject ans = new JSONObject();try {if(application.getAttribute("MQ_MESSAGE") == null) {application.setAttribute("MQ_MESSAGE", new ArrayList<Object>());}List<Object> data = (List<Object>) application.getAttribute("MQ_MESSAGE");ans.put("code", ReturnCode.SUCCESS.getValue());ans.put("data", CloneUtils.clone(data));data.clear();} catch(Exception e) {e.printStackTrace();ans.put("code", ReturnCode.FAILURE.getValue());ans.put("msg", e.getMessage().matches(ReturnCode.REGEX_CHINESE.getValue()) ? e.getMessage() : "內部異常");}return ans;} } View Code

　　spring公共配置文件中<beans/>屬相中加入如下配置。

default-lazy-init="false"

　　6.生產者web1工程

package com.hjz.rabbit.web;import java.util.UUID;import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody;import com.alibaba.fastjson.JSONObject; import com.hjz.mq.common.MqConfig; import com.hjz.mq.common.MqMessage; import com.hjz.mq.sender.MsgSender; import com.hjz.mq.util.ReturnCode;@Controller @RequestMapping("producer") public class MQProducer {@Autowiredprivate MsgSender mqSender;@RequestMapping("process")@ResponseBodypublic JSONObject producer() {JSONObject ans = new JSONObject();try {String msg = "隨機消息 " + UUID.randomUUID().toString();MqMessage message = new MqMessage();message.setBody(msg);mqSender.sendMessage(message, MqConfig.MQ_QUEUE_ADD);ans.put("code", ReturnCode.SUCCESS.getValue());ans.put("data", msg);} catch(Exception e) {ans.put("code", ReturnCode.FAILURE.getValue());ans.put("msg", e.getMessage().matches(ReturnCode.REGEX_CHINESE.getValue()) ? e.getMessage() : "內部錯誤");}return ans;} } View Code

　　7.簡單模擬MQ消息發送和接收web工程

<!DOCTYPE html> <html> <head> <meta http-equiv="Content-Type" content="text/html; charset=UTF-8"> <title>消息展示</title> <script src="/web/scripts/boot.js" type="text/javascript"></script> <link href="/web/scripts/miniui/themes/default/large-mode.css"rel="stylesheet" type="text/css" /> <link href="/web/scripts/miniui/themes/bootstrap/skin.css"rel="stylesheet" type="text/css" /><style> .container {padding: 30px; }.container .mini-panel {margin-right: 20px;margin-bottom: 20px; } </style> </head> <body><div class="container"><div class="mini-clearfix "><div class="mini-col-6"><div class="mini-panel" title="發送消息" width="auto" height="auto"showCollapseButton="true" showCloseButton="false"><div id="panel_send" class="mini-panel" title="模擬消息發送"iconCls="icon-upload" style="width: 600px; height: 300px;"showToolbar="false" showCollapseButton="true" showFooter="false"allowResize="true" collapseOnTitleClick="true"></div></div></div><div class="mini-col-6"><div class="mini-panel mini-panel-primary" title="接收消息" width="auto"showCollapseButton="true" showCloseButton="false"><div id="panel_receive" class="mini-panel mini-panel-primary"title="模擬消息接收" iconCls="icon-download"style="width: 600px; height: 300px;" showToolbar="false"showCollapseButton="true" showFooter="false" allowResize="true"collapseOnTitleClick="true"></div></div></div></div></div> </body> <script type="text/javascript">mini.parse();var panel_send = mini.get('panel_send');var panel_receive = mini.get('panel_receive');$(function() {var myInterval = setInterval(function() {$.ajax({type : "GET",url : "http://localhost:8080/web1/producer/process",//data : "{}",//contentType : "application/json; charset=utf-8", dataType : "json",success : function(data) {if(data.code == 'success') {$(panel_send.getBodyEl()).append('<p>' + data.data + '</p>');} else {report(data.msg);}},error : function(msg) {report(msg);}});$.ajax({type : "GET",url : "http://localhost:8080/web2/consumer/process",dataType : "json",success : function(data) {if(data.code == 'success') {var div = $('<div></div>');data.data.forEach(function(value) {div.append('<p>' + value + '</p>');});$(panel_receive.getBodyEl()).append(div);} else {report(data.msg);}},error : function(msg) {report(msg);}});}, 3000);function report(msg) {clearInterval(myInterval);mini.showMessageBox({showModal: false,title: "提示",iconCls: "mini-messagebox-warning",message: typeof msg === 'object' ? JSON.stringify(msg) : msg,timeout: 3000,});}}) </script> </html> View Code

　　8.tomcat下啟動demo2中的web，web1，web2三個工程。瀏覽器中輸入：http://127.0.0.1:port/web，參看 MQ 模擬示例。

四、完整項目下載地址

　　https://github.com/hjzgg/Rabbit-MQ-demo

轉載于:https://www.cnblogs.com/hujunzheng/p/6321860.html

總結

以上是生活随笔為你收集整理的RabbitMQ安装和使用（和Spring集成）的全部內容，希望文章能夠幫你解決所遇到的問題。

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