如何在SpringMVC项目中使用rabbitmq(rabbitmq,springmvc,开发技术)

1.添加maven依赖

<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配置--><importresource="/application-mq.xml"/>

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

#rabbitMQ配置mq.host=localhostmq.username=donghaomq.password=donghaomq.port=5672mq.vhost=testMQ

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

<beansxmlns="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/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd"><description>rabbitmq连接服务配置</description><!--连接配置--><rabbit:connection-factoryid="connectionFactory"host="${mq.host}"username="${mq.username}"password="${mq.password}"port="${mq.port}"virtual-host="${mq.vhost}"/><rabbit:adminconnection-factory="connectionFactory"/><!--springtemplate声明--><rabbit:templateexchange="koms"id="amqpTemplate"connection-factory="connectionFactory"message-converter="jsonMessageConverter"/><!--消息对象json转换类--><beanid="jsonMessageConverter"class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter"/><!--durable:是否持久化exclusive:仅创建者可以使用的私有队列，断开后自动删除auto_delete:当所有消费客户端连接断开后，是否自动删除队列--><!--申明一个消息队列Queue--><rabbit:queueid="order"name="order"durable="true"auto-delete="false"exclusive="false"/><rabbit:queueid="activity"name="activity"durable="true"auto-delete="false"exclusive="false"/><rabbit:queueid="mail"name="mail"durable="true"auto-delete="false"exclusive="false"/><rabbit:queueid="stock"name="stock"durable="true"auto-delete="false"exclusive="false"/><rabbit:queueid="autoPrint"name="autoPrint"durable="true"auto-delete="false"exclusive="false"/><!--rabbit:direct-exchange：定义exchange模式为direct，意思就是消息与一个特定的路由键完全匹配，才会转发。rabbit:binding：设置消息queue匹配的key--><!--交换机定义--><rabbit:direct-exchangename="koms"durable="true"auto-delete="false"id="koms"><rabbit:bindings><rabbit:bindingqueue="order"key="order"/><rabbit:bindingqueue="activity"key="activity"/><rabbit:bindingqueue="mail"key="mail"/><rabbit:bindingqueue="stock"key="stock"/><rabbit:bindingqueue="autoPrint"key="autoPrint"/></rabbit:bindings></rabbit:direct-exchange><!--queues：监听的队列，多个的话用逗号（,）分隔ref：监听器--><!--配置监听acknowledeg="manual"设置手动应答当消息处理失败时:会一直重发直到消息处理成功--><rabbit:listener-containerconnection-factory="connectionFactory"acknowledge="manual"><!--配置监听器--><rabbit:listenerqueues="activity"ref="activityListener"/><rabbit:listenerqueues="order"ref="orderListener"/><rabbit:listenerqueues="mail"ref="mailListener"/><rabbit:listenerqueues="stock"ref="stockListener"/><rabbit:listenerqueues="autoPrint"ref="autoPrintListener"/></rabbit:listener-container></beans>

5.新增公共入队类

@ServicepublicclassMQProducerImpl{@ResourceprivateAmqpTemplateamqpTemplate;privatefinalstaticLoggerlogger=LoggerFactory.getLogger(MQProducerImpl.class);//公共入队方法publicvoidsendDataToQueue(StringqueueKey,Objectobject){try{amqpTemplate.convertAndSend(queueKey,object);}catch(Exceptione){logger.error(e.toString());}}}

6.创建监听类

importjava.io.IOException;importjava.util.List;importjavax.annotation.Resource;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.core.ChannelAwareMessageListener;importorg.springframework.amqp.utils.SerializationUtils;importorg.springframework.stereotype.Component;importorg.springframework.transaction.annotation.Transactional;importcom.cn.framework.domain.BaseDto;importcom.cn.framework.util.ConstantUtils;importcom.cn.framework.util.RabbitMq.producer.MQProducer;importcom.kxs.service.activityService.IActivityService;importcom.kxs.service.messageService.IMessageService;importcom.rabbitmq.client.Channel;/***活动处理listener*@author*@date2023年6月30日**/@ComponentpublicclassActivityListenerimplementsChannelAwareMessageListener{privatestaticfinalLoggerlog=LoggerFactory.getLogger(ActivityListener.class);@Override@TransactionalpublicvoidonMessage(Messagemessage,Channelchannel){}}

项目启动后 控制台会打印出监听的日志信息 这里写图片描述

结尾：仅供参考，自己用作学习记录，不喜勿喷，共勉！

补充：RabbitMQ与SpringMVC集成并实现发送消息和接收消息（持久化）方案

RabbitMQ本篇不介绍了，直接描述RabbitMQ与SpringMVC集成并实现发送消息和接收消息（持久化）。

使用了Spring-rabbit 发送消息和接收消息，我们使用的Maven来管理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.实现生产者

第一步：是要设置调用安装RabbitMQ的IP、端口等

配置一个global.properties文件

第二步：通过SpringMVC把global.properties文件读进来

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

第三步：配置 RabbitMQ服务器连接、创建rabbitTemplate 消息模板类等，在SpringMVC的配置文件加入下面这些

<beanid="rmqProducer2"class="cn.test.spring.rabbitmq.RmqProducer"></bean><span><!--创建连接类--><beanid="connectionFactory"class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><constructor-argvalue="localhost"/><propertyname="username"value="${rmq.manager.user}"/><propertyname="password"value="${rmq.manager.password}"/><propertyname="host"value="${rmq.ip}"/><propertyname="port"value="${rmq.port}"/></bean><beanid="rabbitAdmin"class="org.springframework.amqp.rabbit.core.RabbitAdmin"><constructor-argref="connectionFactory"/></bean><!--创建rabbitTemplate消息模板类--><beanid="rabbitTemplate"class="org.springframework.amqp.rabbit.core.RabbitTemplate"><constructor-argref="connectionFactory"></constructor-arg></bean></span>

第四步：实现消息类实体和发送消息

类实体

<span>/***消息**/publicclassRabbitMessageimplementsSerializable{ privatestaticfinallongserialVersionUID=-6487839157908352120L; privateClass<?>[]paramTypes;//参数类型 privateStringexchange;//交换器 privateObject[]params; privateStringrouteKey;//路由key publicRabbitMessage(){} publicRabbitMessage(Stringexchange,StringrouteKey,Object...params) { this.params=params; this.exchange=exchange; this.routeKey=routeKey; } @SuppressWarnings("rawtypes") publicRabbitMessage(Stringexchange,StringrouteKey,StringmethodName,Object...params) { this.params=params; this.exchange=exchange; this.routeKey=routeKey; intlen=params.length; Class[]clazzArray=newClass[len]; for(inti=0;i<len;i++) clazzArray[i]=params[i].getClass(); this.paramTypes=clazzArray; } publicbyte[]getSerialBytes() { byte[]res=newbyte[0]; ByteArrayOutputStreambaos=newByteArrayOutputStream(); ObjectOutputStreamoos; try{ oos=newObjectOutputStream(baos); oos.writeObject(this); oos.close(); res=baos.toByteArray(); }catch(IOExceptione){ e.printStackTrace(); } returnres; } publicStringgetRouteKey(){ returnrouteKey; } publicStringgetExchange(){ returnexchange; } publicvoidsetExchange(Stringexchange){ this.exchange=exchange; } publicvoidsetRouteKey(StringrouteKey){ this.routeKey=routeKey; } publicClass<?>[]getParamTypes(){ returnparamTypes; } publicObject[]getParams(){ returnparams; } }</span>

发送消息

<span>/***生产着**/publicclassRmqProducer{@ResourceprivateRabbitTemplaterabbitTemplate;/***发送信息*@parammsg*/publicvoidsendMessage(RabbitMessagemsg){try{System.out.println(rabbitTemplate.getConnectionFactory().getHost());System.out.println(rabbitTemplate.getConnectionFactory().getPort());//发送信息rabbitTemplate.convertAndSend(msg.getExchange(),msg.getRouteKey(),msg);}catch(Exceptione){}}}</span>

说明：

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

源代码中的send调用的方法，一些发送消息帮我们实现好了。

2.上面的代码实现没申明交换器和队列，RabbitMQ不知交换器和队列他们的绑定关系，如果RabbitMQ管理器上没有对应的交换器和队列是不会新建的和关联的，需要手动关联。

我们也可以用代码申明：

rabbitAdmin要申明：eclareExchange方法 参数是交换器

BindingBuilder.bind(queue).to(directExchange).with(queueName);//将queue绑定到exchangerabbitAdmin.declareBinding(binding);//声明绑定关系

源代码有这些方法：

这样就可以实现交换器和队列的绑定关系

交换器我们可以申明为持久化，还有使用完不会自动删除

TopicExchange 参数的说明：name是交换器名称，durable:true 是持久化 autoDelete：false使用完不删除

源代码：

队列也可以申明为持久化

第五步：实现测试类

<span>@ResourceprivateRmqProducerrmqProducer2;@Testpublicvoidtest()throwsIOException{Stringexchange="testExchange";交换器StringrouteKey="testQueue";//队列StringmethodName="test";//调用的方法//参数Map<String,Object>param=newHashMap<String,Object>();param.put("data","hello");RabbitMessagemsg=newRabbitMessage(exchange,routeKey,methodName,param);//发送消息rmqProducer2.sendMessage(msg);}</span>

结果：RabbitMQ有一条消息

2.消费者

第一步：RabbitMQ服务器连接这些在生产者那边已经介绍了，这边就不介绍了，我们要配置 RabbitMQ服务器连接、创建rabbitTemplate 消息模板类、消息转换器、消息转换器监听器等，在SpringMVC的配置文件加入下面这些

<span><!--创建连接类--><beanid="connectionFactory"class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><constructor-argvalue="localhost"/><propertyname="username"value="${rmq.manager.user}"/><propertyname="password"value="${rmq.manager.password}"/><propertyname="host"value="${rmq.ip}"/><propertyname="port"value="${rmq.port}"/></bean><beanid="rabbitAdmin"class="org.springframework.amqp.rabbit.core.RabbitAdmin"><constructor-argref="connectionFactory"/></bean><!--创建rabbitTemplate消息模板类--><beanid="rabbitTemplate"class="org.springframework.amqp.rabbit.core.RabbitTemplate"><constructor-argref="connectionFactory"></constructor-arg></bean><!--创建消息转换器为SimpleMessageConverter--><beanid="serializerMessageConverter"class="org.springframework.amqp.support.converter.SimpleMessageConverter"></bean><!--设置持久化的队列--><beanid="queue"class="org.springframework.amqp.core.Queue"><constructor-argindex="0"value="testQueue"></constructor-arg><constructor-argindex="1"value="true"></constructor-arg><constructor-argindex="2"value="false"></constructor-arg><constructor-argindex="3"value="false"></constructor-arg></bean><!--创建交换器的类型并持久化--><beanid="topicExchange"class="org.springframework.amqp.core.TopicExchange"><constructor-argindex="0"value="testExchange"></constructor-arg><constructor-argindex="1"value="true"></constructor-arg><constructor-argindex="2"value="false"></constructor-arg></bean><util:mapid="arguments"></util:map><!--绑定交换器、队列--><beanid="binding"class="org.springframework.amqp.core.Binding"><constructor-argindex="0"value="testQueue"></constructor-arg><constructor-argindex="1"value="QUEUE"></constructor-arg><constructor-argindex="2"value="testExchange"></constructor-arg><constructor-argindex="3"value="testQueue"></constructor-arg><constructor-argindex="4"value="#{arguments}"></constructor-arg></bean><!--用于接收消息的处理类--><beanid="rmqConsumer"class="cn.test.spring.rabbitmq.RmqConsumer"></bean><beanid="messageListenerAdapter"class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter"><constructor-argref="rmqConsumer"/><propertyname="defaultListenerMethod"value="rmqProducerMessage"></property><propertyname="messageConverter"ref="serializerMessageConverter"></property></bean><!--用于消息的监听的容器类SimpleMessageListenerContainer,监听队列queues可以传多个--><beanid="listenerContainer"class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer"><propertyname="queues"ref="queue"></property><propertyname="connectionFactory"ref="connectionFactory"></property><propertyname="messageListener"ref="messageListenerAdapter"></property></bean></span>

说明：

1.org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer中的queues可以传入多个队列

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

有哪个消费者适配器来处理 ，参数defaultListenerMethod是默认调用方法来处理消息。

3.交换器和队列的持久化在生产者有介绍过了。

4.org.springframework.amqp.core.Binding这个类的绑定，在SpringMVC配置文件中配置时，

DestinationType这个参数要注意点

源代码：

第二步：处理消息

<span>/***消费者**/publicclassRmqConsumer{publicvoidrmqProducerMessage(Objectobject){RabbitMessagerabbitMessage=(RabbitMessage)object;System.out.println(rabbitMessage.getExchange());System.out.println(rabbitMessage.getRouteKey());System.out.println(rabbitMessage.getParams().toString());}}</span>

在启动过程中会报这样的错误，可能是你的交换器和队列没配置好