Spring使用Rabbitmq (简单使用)

我不是女神ヾ 2022-01-07 08:11 380阅读 0赞

1、pom.xml jar包引用

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1   <dependencies>
     2         <dependency>
     3             <groupId>org.springframework</groupId>
     4             <artifactId>spring-test</artifactId>
     5         </dependency>
     6         <dependency>
     7             <groupId>org.springframework</groupId>
     8             <artifactId>spring-webmvc</artifactId>
     9         </dependency>
    10         
    11         <dependency>
    12             <groupId>org.springframework.amqp</groupId>
    13             <artifactId>spring-amqp</artifactId>
    14             <version>2.1.2.RELEASE</version>
    15         </dependency>
    16         <dependency>
    17             <groupId>org.springframework.amqp</groupId>
    18             <artifactId>spring-rabbit</artifactId>
    19             <version>2.1.2.RELEASE</version>
    20         </dependency>
    21         
    22         <dependency>
    23             <groupId>com.rabbitmq</groupId>
    24             <artifactId>amqp-client</artifactId>
    25             <version>5.6.0</version>
    26             <exclusions>
    27                 <exclusion>
    28                     <groupId>org.slf4j</groupId>
    29                     <artifactId>slf4j-api</artifactId>
    30                 </exclusion>
    31             </exclusions>
    32         </dependency>
    33         
    34         <dependency>
    35             <groupId>org.slf4j</groupId>
    36             <artifactId>slf4j-api</artifactId>
    37         </dependency>
    38         <dependency>
    39             <groupId>org.slf4j</groupId>
    40             <artifactId>jcl-over-slf4j</artifactId>
    41         </dependency>
    42         <dependency>
    43             <groupId>org.apache.logging.log4j</groupId>
    44             <artifactId>log4j-api</artifactId>
    45         </dependency>
    46         <dependency>
    47             <groupId>org.apache.logging.log4j</groupId>
    48             <artifactId>log4j-core</artifactId>
    49         </dependency>
    50         <dependency>
    51             <groupId>org.apache.logging.log4j</groupId>
    52             <artifactId>log4j-1.2-api</artifactId>
    53         </dependency>
    54         <dependency>
    55             <groupId>org.apache.logging.log4j</groupId>
    56             <artifactId>log4j-jcl</artifactId>
    57         </dependency>
    58         <dependency>
    59             <groupId>org.apache.logging.log4j</groupId>
    60             <artifactId>log4j-slf4j-impl</artifactId>
    61         </dependency>
    62         <dependency>
    63             <groupId>org.apache.logging.log4j</groupId>
    64             <artifactId>log4j-web</artifactId>
    65         </dependency>
    66     </dependencies>

2、resouces 下面的log4j2.xml

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 <?xml version="1.0" encoding="UTF-8"?>
     2 <Configuration status="info" monitorInterval="30">
     3     
     4     <appenders>
     5         
     6         <Console name="Console" target="SYSTEM_OUT">
     7             
     8             <ThresholdFilter level="debug" onMatch="ACCEPT" onMismatch="DENY" />
     9             
    10             <PatternLayout pattern="[%-5p] %d{yyyy-MM-dd HH:mm:ss} method:%l%n%m%n"/>
    11         </Console>
    12     </appenders>
    13     
    14     <loggers>
    15         
    16         <root level="info">
    17             <appender-ref ref="Console" />
    18         </root>
    19     </loggers>
    20 </Configuration>

3、resouces 下面的spring/rabbitmq-context.xml

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 <?xml version="1.0" encoding="UTF-8"?>
     2 <beans xmlns="http://www.springframework.org/schema/beans"
     3        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
     4        xmlns:context="http://www.springframework.org/schema/context"
     5        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
     6     <context:component-scan base-package="com.tandaima.rabbitmq.spring"/>
     7     
     8     <context:property-placeholder file-encoding="utf-8" location="classpath:rabbitmq.properties"/>
     9     <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
    10         <property name="host" value="${mq.host}" />
    11         <property name="port" value="${mq.port}" />
    12         <property name="username" value="${mq.username}" />
    13         <property name="password" value="${mq.password}" />
    14         <property name="virtualHost" value="${mq.virtual-host}" />
    15         
    20         <property name="cacheMode" value="CHANNEL"/>
    21         
    23         <property name="channelCacheSize" value="10"/>
    24         
    25         
    26         <property name="channelCheckoutTimeout" value="200"/>
    27         
    28         <property name="publisherConfirms" value="true"/>
    29     </bean>
    30     
    31     <bean id="rabbitTemplate"  class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    32         <constructor-arg name="connectionFactory" ref="connectionFactory"/>
    33         
    34         <property name="confirmCallback" ref="confirmCallback"/>
    35         
    36         <property name="returnCallback" ref="returnCallback"/>
    37     </bean>
    38     
    39     <rabbit:admin id="connectAdmin" connection-factory="connectionFactory" />
    40 
    41     
    42     <bean id="confirmCallback" class="com.tandaima.rabbitmq.spring.service.ConfirmCallBackListener"/>
    43     
    44     <bean id="returnCallback" class="com.tandaima.rabbitmq.spring.service.ReturnCallBackListener"/>
    45     
    49 
    50     
    51     <bean id="transactionRecordMessage" class="com.tandaima.rabbitmq.spring.receive.TransactionRecordMessage"/>
    52     <rabbit:queue name="transaction.record.queue" durable="true" exclusive="false" auto-delete="false"/>
    53     
    54     <rabbit:direct-exchange name="${mq.exchange-transaction}" declared-by="connectAdmin" durable="true">
    55         <rabbit:bindings>
    56             <rabbit:binding queue="transaction.record.queue" key="${mq.routing-key-transaction-record}"/>
    57         </rabbit:bindings>
    58     </rabbit:direct-exchange>
    59     
    60     
    61     <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
    62         <rabbit:listener queues="transaction.record.queue" ref="transactionRecordMessage" />
    63     </rabbit:listener-container>
    64 
    65     
    66     <bean id="transactionKlineMessage" class="com.tandaima.rabbitmq.spring.receive.TransactionKlineMessage"/>
    67     <rabbit:queue name="transaction.kline.queue" durable="true" exclusive="false" auto-delete="false"/>
    68     <rabbit:direct-exchange name="${mq.exchange-transaction}" declared-by="connectAdmin" durable="true">
    69         <rabbit:bindings>
    70             <rabbit:binding queue="transaction.kline.queue" key="${mq.routing-key-transaction-kline}"/>
    71         </rabbit:bindings>
    72     </rabbit:direct-exchange>
    73     <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
    74         <rabbit:listener queues="transaction.kline.queue" ref="transactionKlineMessage" />
    75     </rabbit:listener-container>
    76 </beans>

4、resouces 下面的rabbitmq.properties

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 mq.host=127.0.0.1
     2 mq.username=lpz
     3 mq.password=lpz
     4 mq.port=5672
     5 mq.virtual-host=/
     6 #交易记录队列key
     7 mq.routing-key-transaction-record=transaction.record.queue.key
     8 #交易K线队列key
     9 mq.routing-key-transaction-kline=transaction.kline.queue.key
    10 #交易相关交换机
    11 mq.exchange-transaction=transaction.exchange

5、定义消息发送类MessageProducer

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 package com.tandaima.rabbitmq.spring.service;
     2 
     3 import org.apache.log4j.Logger;
     4 import org.springframework.amqp.core.AmqpTemplate;
     5 import org.springframework.beans.factory.annotation.Autowired;
     6 import org.springframework.stereotype.Service;
     7 
     8 @Service
     9 public class MessageProducer {
    10 
    11     private Logger logger= Logger.getLogger(MessageProducer.class);
    12 
    13     @Autowired
    14     private AmqpTemplate amqpTemplate;
    15 
    16     /**
    17      * 发送消息到队列
    18      * @param exchange 通道名称
    19      * @param queueKey 队列key
    20      * @param content 内容
    21      */
    22     public void sendMessage(String exchange,String queueKey,Object content){
    23         try {
    24             amqpTemplate.convertAndSend(exchange,queueKey, content);
    25         }catch (Exception e){
    26             logger.error("RabbitMQ发送消息异常==>"+e.getMessage());
    27         }
    28     }
    29 }

6、消息发送成功回调类ConfirmCallBackListener

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 package com.tandaima.rabbitmq.spring.service;
     2 import org.apache.log4j.Logger;
     3 import org.springframework.amqp.rabbit.connection.CorrelationData;
     4 import org.springframework.amqp.rabbit.core.RabbitTemplate;
     5 import org.springframework.context.annotation.Configuration;
     6 
     7 @Configuration
     8 public class ConfirmCallBackListener implements RabbitTemplate.ConfirmCallback{
     9     private Logger logger=Logger.getLogger(ConfirmCallBackListener.class);
    10     /**
    11      * CorrelationData 是在发送消息时传入回调方法的参数，可以用于区分消息对象。 CorrelationData对象中只有一个属性 String id。
    12      * 通过这个参数，我们可以区分当前是发送哪一条消息时的回调，并通过ack参数来进行失败重发功能
    13      *
    14      * @param correlationData 回调的相关数据.
    15      * @param ack true for ack, false for nack
    16      * @param cause 专门给NACK准备的一个可选的原因，其他情况为null。
    17      */
    18     @Override
    19     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    20         logger.info("exchange确认"+ack);
    21     }
    22 }

7、消息发送失败回调类ReturnCallBackListener

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 package com.tandaima.rabbitmq.spring.service;
     2 
     3 import org.apache.log4j.Logger;
     4 import org.springframework.amqp.core.Message;
     5 import org.springframework.amqp.rabbit.core.RabbitTemplate;
     6 import org.springframework.context.annotation.Configuration;
     7 
     8 @Configuration
     9 public class ReturnCallBackListener implements RabbitTemplate.ReturnCallback{
    10     private Logger logger=Logger.getLogger(ReturnCallBackListener.class);
    11     @Override
    12     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
    13         logger.error("失败确认:"+message+" | "+replyCode+" | "+replyText+" | "+exchange+" | "+routingKey);
    14     }
    15 }

8、定义TransactionKlineMessage

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 package com.tandaima.rabbitmq.spring.receive;
     2 
     3 import com.alibaba.fastjson.JSON;
     4 import com.rabbitmq.client.Channel;
     5 import org.apache.log4j.Logger;
     6 import org.springframework.amqp.core.Message;
     7 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
     8 import org.springframework.context.annotation.Configuration;
     9 
    10 @Configuration
    11 public class TransactionKlineMessage implements ChannelAwareMessageListener {
    12     private Logger logger=Logger.getLogger(TransactionKlineMessage.class);
    13     @Override
    14     public void onMessage(Message message, Channel channel) throws Exception {
    15         try{
    16             String str = new String(message.getBody());
    17             logger.info("接收到交易K线信息==>:" + str);
    18             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    19         }catch(Exception e){
    20             logger.error("接收到交易K线信息异常回滚消息到队列中==>"+e.getMessage());
    21             //消息回滚通道
    22             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
    23         }
    24     }
    25 
    26     @Override
    27     public void onMessage(Message message) {
    28         logger.info("消息==>"+JSON.toJSONString(message));
    29     }
    30 }

9、定义TransactionRecordMessage

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 package com.tandaima.rabbitmq.spring.receive;
     2 
     3 import com.alibaba.fastjson.JSON;
     4 import com.rabbitmq.client.Channel;
     5 import org.apache.log4j.Logger;
     6 import org.springframework.amqp.core.Message;
     7 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
     8 import org.springframework.context.annotation.Configuration;
     9 
    10 @Configuration
    11 public class TransactionRecordMessage implements ChannelAwareMessageListener {
    12     private Logger logger=Logger.getLogger(TransactionRecordMessage.class);
    13     @Override
    14     public void onMessage(Message message, Channel channel) throws Exception {
    15         try{
    16             String str = new String(message.getBody());
    17             logger.info("接收到交易记录信息==>:" + str);
    18             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    19         }catch(Exception e){
    20             logger.error("接收到交易记录信息异常回滚消息到队列中==>"+e.getMessage());
    21             //消息回滚通道
    22             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
    23         }
    24     }
    25 
    26     @Override
    27     public void onMessage(Message message) {
    28         logger.info("消息==>"+JSON.toJSONString(message));
    29     }
    30 }

10、测试类RabbitmqMessageTest

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

1 package com.tandaima.rabbitmq.spring;
     2 
     3 import com.tandaima.rabbitmq.spring.service.MessageProducer;
     4 import org.junit.Test;
     5 import org.junit.runner.RunWith;
     6 import org.springframework.beans.factory.annotation.Autowired;
     7 import org.springframework.test.context.ContextConfiguration;
     8 import org.springframework.test.context.junit4.SpringRunner;
     9 
    10 import java.util.Date;
    11 
    12 @RunWith(SpringRunner.class)
    13 @ContextConfiguration(locations = {"classpath:spring/rabbitmq-context.xml"})
    14 public class RabbitmqMessageTest {
    15     @Autowired
    16     private MessageProducer messageProducer;
    17 
    18     private String exchange="transaction.exchange";
    19 
    20     @Test
    21     public void senTransactionRecordMsg(){
    22         for(int i=1;i<=5;i++){
    23             String routingKeyTransactionRecord = "transaction.record.queue.key";
    24             messageProducer.sendMessage(
    25                     exchange,
    26                     routingKeyTransactionRecord,
    27                     "我是内容"+ i);
    28         }
    29     }
    30 
    31     /**
    32      * 10w条数据 发送38秒496毫秒
    33      */
    34     @Test
    35     public void senTransactionKlineMsg(){
    36         Date begin=new Date(); //开始时间
    37         for(int i=1;i<=10;i++){
    38             String routingKeyTransactionKlin = "transaction.kline.queue.key";
    39             messageProducer.sendMessage(
    40                     exchange,
    41                     routingKeyTransactionKlin,
    42                     "我是内容"+ i);
    43         }
    44         System.out.println(getString(begin,new Date()));
    45     }
    46     private String getString(Date begin,Date end){
    47         long  between = end.getTime() - begin.getTime();// 得到两者的毫秒数
    48         long day = between / (24 * 60 * 60 * 1000);
    49         long hour = (between / (60 * 60 * 1000) - day * 24);
    50         long min = ((between / (60 * 1000)) - day * 24 * 60 - hour * 60);
    51         long s = (between / 1000 - day * 24 * 60 * 60 - hour * 60 * 60 - min * 60);
    52         long ms = (between - day * 24 * 60 * 60 * 1000 - hour * 60 * 60 * 1000 - min * 60 * 1000 - s * 1000);
    53         return (day + "天" + hour + "小时" + min + "分" + s + "秒" + ms  + "毫秒");
    54     }
    55 }

测试效果

![1329419-20190620095828911-1860687968.png][]

我这里用的是路由模式交换机、两个队列使用一个交换机

转载于:https://www.cnblogs.com/longpizi/p/11056755.html

[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20211227/77b0d9432d82453bb8b4fc92450ac032.png
[1329419-20190620095828911-1860687968.png]: /images/20211227/1e8c40dbafb041c0b02a5924b36365fc.png