溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶(hù)服務(wù)條款》

如何在SpringMVC項(xiàng)目中使用rabbitmq

發(fā)布時(shí)間：2021-01-20 14:13:14 來(lái)源：億速云 閱讀：901 作者：Leah 欄目：開(kāi)發(fā)技術(shù)

這篇文章給大家介紹如何在SpringMVC項(xiàng)目中使用rabbitmq，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

1.添加maven依賴(lài)

<dependency>
 <groupId>com.rabbitmq</groupId>
 <artifactId>amqp-client</artifactId>
 <version>3.5.1</version>
</dependency>
<dependency>
 <groupId>org.springframework.amqp</groupId>
 <artifactId>spring-rabbit</artifactId>
 <version>1.4.5.RELEASE</version>
</dependency>

2.spring主配置文件中加入rabbitMQ xml文件的配置

<!-- rabbitMQ 配置 -->
 <import resource="/application-mq.xml"/>

3.jdbc配置文件中加入 rabbitmq的鏈接配置

#rabbitMQ配置
mq.host=localhost
mq.username=donghao
mq.password=donghao
mq.port=5672
mq.vhost=testMQ

4.新建application-mq.xml文件,添加配置信息

 <beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xmlns:rabbit="http://www.springframework.org/schema/rabbit" 
 xsi:schemaLocation="http://www.springframework.org/schema/beans 
 http://www.springframework.org/schema/beans/spring-beans-3.0.xsd 
 http://www.springframework.org/schema/rabbit
 http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" >
 <description>rabbitmq 連接服務(wù)配置</description>
 <!-- 連接配置 -->
 <rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}" virtual-host="${mq.vhost}"/>
 <rabbit:admin connection-factory="connectionFactory"/>
 <!-- spring template聲明-->
 <rabbit:template exchange="koms" id="amqpTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
 <!-- 消息對(duì)象json轉(zhuǎn)換類(lèi) -->
 <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> 
 <!-- 
  durable:是否持久化
  exclusive: 僅創(chuàng)建者可以使用的私有隊(duì)列，斷開(kāi)后自動(dòng)刪除
  auto_delete: 當(dāng)所有消費(fèi)客戶(hù)端連接斷開(kāi)后，是否自動(dòng)刪除隊(duì)列
  -->
  <!-- 申明一個(gè)消息隊(duì)列Queue -->
 <rabbit:queue id="order" name="order" durable="true" auto-delete="false" exclusive="false" />
  <rabbit:queue id="activity" name="activity" durable="true" auto-delete="false" exclusive="false" />
  <rabbit:queue id="mail" name="mail" durable="true" auto-delete="false" exclusive="false" />
  <rabbit:queue id="stock" name="stock" durable="true" auto-delete="false" exclusive="false" />
  <rabbit:queue id="autoPrint" name="autoPrint" durable="true" auto-delete="false" exclusive="false" />
 <!--
  rabbit:direct-exchange：定義exchange模式為direct，意思就是消息與一個(gè)特定的路由鍵完全匹配，才會(huì)轉(zhuǎn)發(fā)。 
 rabbit:binding：設(shè)置消息queue匹配的key
  -->
 <!-- 交換機(jī)定義 -->
 <rabbit:direct-exchange name="koms" durable="true" auto-delete="false" id="koms">
 <rabbit:bindings>
  <rabbit:binding queue="order" key="order"/>
   <rabbit:binding queue="activity" key="activity"/>
   <rabbit:binding queue="mail" key="mail"/>
   <rabbit:binding queue="stock" key="stock"/>
   <rabbit:binding queue="autoPrint" key="autoPrint"/>
 </rabbit:bindings>
</rabbit:direct-exchange>
 <!--
   queues：監(jiān)聽(tīng)的隊(duì)列，多個(gè)的話(huà)用逗號(hào)（,）分隔 
  ref：監(jiān)聽(tīng)器
  -->
 <!-- 配置監(jiān)聽(tīng) acknowledeg = "manual" 設(shè)置手動(dòng)應(yīng)答 當(dāng)消息處理失敗時(shí):會(huì)一直重發(fā) 直到消息處理成功 -->
 <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
 <!-- 配置監(jiān)聽(tīng)器 -->
  <rabbit:listener queues="activity" ref="activityListener"/>
   <rabbit:listener queues="order" ref="orderListener"/>
  <rabbit:listener queues="mail" ref="mailListener"/>
  <rabbit:listener queues="stock" ref="stockListener"/>
  <rabbit:listener queues="autoPrint" ref="autoPrintListener"/>
 </rabbit:listener-container>
</beans>

5.新增公共入隊(duì)類(lèi)

@Service
public class MQProducerImpl{
@Resource
 private AmqpTemplate amqpTemplate;
 private final static Logger logger = LoggerFactory.getLogger(MQProducerImpl.class);
 //公共入隊(duì)方法
 public void sendDataToQueue(String queueKey, Object object) {
  try {
   amqpTemplate.convertAndSend(queueKey, object);
  } catch (Exception e) {
   logger.error(e.toString());
  }
 }
}

6.創(chuàng)建監(jiān)聽(tīng)類(lèi)

如何在SpringMVC項(xiàng)目中使用rabbitmq

import java.io.IOException;
import java.util.List;
import javax.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import com.cn.framework.domain.BaseDto;
import com.cn.framework.util.ConstantUtils;
import com.cn.framework.util.RabbitMq.producer.MQProducer;
import com.kxs.service.activityService.IActivityService;
import com.kxs.service.messageService.IMessageService;
import com.rabbitmq.client.Channel;
/**
 * 活動(dòng)處理listener
* @author
* @date 2017年6月30日
**/
@Component
public class ActivityListener implements ChannelAwareMessageListener {
 private static final Logger log = LoggerFactory.getLogger(ActivityListener.class);
 @Override
 @Transactional
 public void onMessage(Message message,Channel channel) {
 }
}

如何在SpringMVC項(xiàng)目中使用rabbitmq

項(xiàng)目啟動(dòng)后 控制臺(tái)會(huì)打印出監(jiān)聽(tīng)的日志信息 這里寫(xiě)圖片描述

結(jié)尾：僅供參考，自己用作學(xué)習(xí)記錄，不喜勿噴，共勉！

補(bǔ)充：RabbitMQ與SpringMVC集成并實(shí)現(xiàn)發(fā)送消息和接收消息（持久化）方案

RabbitMQ本篇不介紹了，直接描述RabbitMQ與SpringMVC集成并實(shí)現(xiàn)發(fā)送消息和接收消息（持久化）。

使用了Spring-rabbit 發(fā)送消息和接收消息，我們使用的Maven來(lái)管理Jar包，在Maven的pom.xml文件中引入jar包

<span > <dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit</artifactId>
   <version>1.3.6.RELEASE</version>
 </dependency></span>

1.實(shí)現(xiàn)生產(chǎn)者

第一步：是要設(shè)置調(diào)用安裝RabbitMQ的IP、端口等

配置一個(gè)global.properties文件

如何在SpringMVC項(xiàng)目中使用rabbitmq

第二步：通過(guò)SpringMVC把global.properties文件讀進(jìn)來(lái)

<span ><!-- 注入屬性文件 --> 
 <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> 
  <property name="locations"> 
   <list> 
    <value>classpath:global.properties</value> 
   </list> 
  </property> 
 </bean> </span>

第三步：配置 RabbitMQ服務(wù)器連接、創(chuàng)建rabbitTemplate 消息模板類(lèi)等，在SpringMVC的配置文件加入下面這些

<bean id="rmqProducer2" class="cn.test.spring.rabbitmq.RmqProducer"></bean>
<span > <!-- 創(chuàng)建連接類(lèi) --> 
 <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"> 
  <constructor-arg value="localhost" /> 
  <property name="username" value="${rmq.manager.user}" /> 
  <property name="password" value="${rmq.manager.password}" /> 
  <property name="host" value="${rmq.ip}" /> 
  <property name="port" value="${rmq.port}" /> 
 </bean> 
  
 <bean id="rabbitAdmin" class="org.springframework.amqp.rabbit.core.RabbitAdmin"> 
  <constructor-arg ref="connectionFactory" /> 
 </bean> 
  <!-- 創(chuàng)建rabbitTemplate 消息模板類(lèi) --> 
 <bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"> 
  <constructor-arg ref="connectionFactory"></constructor-arg> 
 </bean> </span>

第四步：實(shí)現(xiàn)消息類(lèi)實(shí)體和發(fā)送消息

類(lèi)實(shí)體

<span >/**
 * 消息
 *
 */
public class RabbitMessage implements Serializable
{
	private static final long serialVersionUID = -6487839157908352120L;	
	private Class<?>[] paramTypes;//參數(shù)類(lèi)型
	private String exchange;//交換器	
	private Object[] params;	
	private String routeKey;//路由key	
	public RabbitMessage(){} 
	public RabbitMessage(String exchange,String routeKey,Object...params)
	{
		this.params=params;		
		this.exchange=exchange;
		this.routeKey=routeKey;
	}
	
	@SuppressWarnings("rawtypes")
	public RabbitMessage(String exchange,String routeKey,String methodName,Object...params)
	{
		this.params=params;		
		this.exchange=exchange;
		this.routeKey=routeKey;
		int len=params.length;
		Class[] clazzArray=new Class[len];
		for(int i=0;i<len;i++)
			clazzArray[i]=params[i].getClass();
		this.paramTypes=clazzArray;
	}
	
	public byte[] getSerialBytes()
	{
		byte[] res=new byte[0];
		ByteArrayOutputStream baos=new ByteArrayOutputStream();
		ObjectOutputStream oos;
		try {
			oos = new ObjectOutputStream(baos);
			oos.writeObject(this);
			oos.close();
			res=baos.toByteArray();
		} catch (IOException e) {
			e.printStackTrace();
		}		
		return res;
	}	
	
	public String getRouteKey() {
		return routeKey;
	} 
 
	public String getExchange() {
		return exchange;
	}
 
	public void setExchange(String exchange) {
		this.exchange = exchange;
	}
 
	public void setRouteKey(String routeKey) {
		this.routeKey = routeKey;
	} 
 
	public Class<?>[] getParamTypes() {
		return paramTypes;
	} 

	public Object[] getParams() {
		return params;
	}	
}
</span>

發(fā)送消息

<span >/**
 * 生產(chǎn)著
 *
 */ 
public class RmqProducer
{ 
 @Resource
 private RabbitTemplate rabbitTemplate; 
 /**
 * 發(fā)送信息
 * @param msg
 */
 public void sendMessage(RabbitMessage msg)
 {
 try {
 System.out.println(rabbitTemplate.getConnectionFactory().getHost());
 System.out.println(rabbitTemplate.getConnectionFactory().getPort());
 //發(fā)送信息
  rabbitTemplate.convertAndSend(msg.getExchange(), msg.getRouteKey(), msg); 
 } catch (Exception e) {
 } 
 } 
}</span>

說(shuō)明：

1. rabbitTemplate.convertAndSend(msg.getExchange(), msg.getRouteKey(), msg);

源代碼中的send調(diào)用的方法，一些發(fā)送消息幫我們實(shí)現(xiàn)好了。

如何在SpringMVC項(xiàng)目中使用rabbitmq

2.上面的代碼實(shí)現(xiàn)沒(méi)申明交換器和隊(duì)列，RabbitMQ不知交換器和隊(duì)列他們的綁定關(guān)系，如果RabbitMQ管理器上沒(méi)有對(duì)應(yīng)的交換器和隊(duì)列是不會(huì)新建的和關(guān)聯(lián)的，需要手動(dòng)關(guān)聯(lián)。

如何在SpringMVC項(xiàng)目中使用rabbitmq

我們也可以用代碼申明：

rabbitAdmin要申明：eclareExchange方法 參數(shù)是交換器

BindingBuilder.bind(queue).to(directExchange).with(queueName);//將queue綁定到exchange 
rabbitAdmin.declareBinding(binding);//聲明綁定關(guān)系

源代碼有這些方法：

如何在SpringMVC項(xiàng)目中使用rabbitmq

這樣就可以實(shí)現(xiàn)交換器和隊(duì)列的綁定關(guān)系

交換器我們可以申明為持久化，還有使用完不會(huì)自動(dòng)刪除

TopicExchange 參數(shù)的說(shuō)明：name是交換器名稱(chēng)，durable:true 是持久化 autoDelete：false使用完不刪除

源代碼：

如何在SpringMVC項(xiàng)目中使用rabbitmq

隊(duì)列也可以申明為持久化

如何在SpringMVC項(xiàng)目中使用rabbitmq

第五步：實(shí)現(xiàn)測(cè)試類(lèi)

<span >@Resource
 private RmqProducer rmqProducer2; 
 @Test
 public void test() throws IOException
 { 
 String exchange="testExchange";交換器
 String routeKey="testQueue";//隊(duì)列
 String methodName="test";//調(diào)用的方法
 //參數(shù)
 Map<String,Object> param=new HashMap<String, Object>();
 param.put("data","hello");
 
 RabbitMessage msg=new RabbitMessage(exchange,routeKey, methodName, param);
 //發(fā)送消息
 rmqProducer2.sendMessage(msg);
 
 }</span>

結(jié)果：RabbitMQ有一條消息

如何在SpringMVC項(xiàng)目中使用rabbitmq

2.消費(fèi)者

第一步：RabbitMQ服務(wù)器連接這些在生產(chǎn)者那邊已經(jīng)介紹了，這邊就不介紹了，我們要配置 RabbitMQ服務(wù)器連接、創(chuàng)建rabbitTemplate 消息模板類(lèi)、消息轉(zhuǎn)換器、消息轉(zhuǎn)換器監(jiān)聽(tīng)器等，在SpringMVC的配置文件加入下面這些

<span > <!-- 創(chuàng)建連接類(lèi) --> 
 <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"> 
  <constructor-arg value="localhost" /> 
  <property name="username" value="${rmq.manager.user}" /> 
  <property name="password" value="${rmq.manager.password}" /> 
  <property name="host" value="${rmq.ip}" /> 
  <property name="port" value="${rmq.port}" /> 
 </bean> 
  
 <bean id="rabbitAdmin" class="org.springframework.amqp.rabbit.core.RabbitAdmin"> 
  <constructor-arg ref="connectionFactory" /> 
 </bean> 
  <!-- 創(chuàng)建rabbitTemplate 消息模板類(lèi) --> 
 <bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"> 
  <constructor-arg ref="connectionFactory"></constructor-arg> 
 </bean>  
 
  <!-- 創(chuàng)建消息轉(zhuǎn)換器為SimpleMessageConverter --> 
 <bean id="serializerMessageConverter" class="org.springframework.amqp.support.converter.SimpleMessageConverter"></bean>  
 
 <!-- 設(shè)置持久化的隊(duì)列 --> 
 <bean id="queue" class="org.springframework.amqp.core.Queue"> 
  <constructor-arg index="0" value="testQueue"></constructor-arg> 
  <constructor-arg index="1" value="true"></constructor-arg> 
  <constructor-arg index="2" value="false"></constructor-arg> 
  <constructor-arg index="3" value="false"></constructor-arg> 
 </bean>  
 
  <!--創(chuàng)建交換器的類(lèi)型 并持久化--> 
 <bean id="topicExchange" class="org.springframework.amqp.core.TopicExchange">
  <constructor-arg index="0" value="testExchange"></constructor-arg> 
  <constructor-arg index="1" value="true"></constructor-arg> 
  <constructor-arg index="2" value="false"></constructor-arg> 
 </bean>
 
 <util:map id="arguments">  
 </util:map> 
 
 <!-- 綁定交換器、隊(duì)列 --> 
 <bean id="binding" class="org.springframework.amqp.core.Binding"> 
  <constructor-arg index="0" value="testQueue"></constructor-arg> 
  <constructor-arg index="1" value="QUEUE"></constructor-arg> 
  <constructor-arg index="2" value="testExchange"></constructor-arg>
  <constructor-arg index="3" value="testQueue"></constructor-arg> 
  <constructor-arg index="4" value="#{arguments}"></constructor-arg> 
 </bean> 
 
 
 <!-- 用于接收消息的處理類(lèi) --> 
 <bean id="rmqConsumer" class="cn.test.spring.rabbitmq.RmqConsumer"></bean> 
 
 <bean id="messageListenerAdapter" class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter"> 
  <constructor-arg ref="rmqConsumer" /> 
  <property name="defaultListenerMethod" value="rmqProducerMessage"></property> 
  <property name="messageConverter" ref="serializerMessageConverter"></property> 
 </bean> 
 
 <!-- 用于消息的監(jiān)聽(tīng)的容器類(lèi)SimpleMessageListenerContainer,監(jiān)聽(tīng)隊(duì)列 queues可以傳多個(gè)--> 
 <bean id="listenerContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer"> 
  <property name="queues" ref="queue"></property> 
  <property name="connectionFactory" ref="connectionFactory"></property> 
  <property name="messageListener" ref="messageListenerAdapter"></property> 
 </bean> 
 </span>

說(shuō)明：

1.org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer中的queues可以傳入多個(gè)隊(duì)列

如何在SpringMVC項(xiàng)目中使用rabbitmq

2.org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter

有哪個(gè)消費(fèi)者適配器來(lái)處理 ，參數(shù)defaultListenerMethod是默認(rèn)調(diào)用方法來(lái)處理消息。

3.交換器和隊(duì)列的持久化在生產(chǎn)者有介紹過(guò)了。

4.org.springframework.amqp.core.Binding這個(gè)類(lèi)的綁定，在SpringMVC配置文件中配置時(shí)，

DestinationType這個(gè)參數(shù)要注意點(diǎn)

源代碼：

如何在SpringMVC項(xiàng)目中使用rabbitmq

第二步：處理消息

<span >/**
 * 消費(fèi)者
 *
 */
public class RmqConsumer 
{
 public void rmqProducerMessage(Object object){ 
 RabbitMessage rabbitMessage=(RabbitMessage) object; 
 System.out.println(rabbitMessage.getExchange());
 System.out.println(rabbitMessage.getRouteKey());
 System.out.println(rabbitMessage.getParams().toString()); 
 } 
}</span>

在啟動(dòng)過(guò)程中會(huì)報(bào)這樣的錯(cuò)誤，可能是你的交換器和隊(duì)列沒(méi)配置好

如何在SpringMVC項(xiàng)目中使用rabbitmq

關(guān)于如何在SpringMVC項(xiàng)目中使用rabbitmq就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。

向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀(guān)點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手