RabbitMQ入门到实战(6)-与spring集成

雨点打透心脏的1/2处 2022-05-27 03:17 117阅读 0赞

转载于：http://www.cnblogs.com/4----/p/6668467.html

# 1.简介 #

原计划这章应该讲RabbitMQ的RPC调用的，后来想想，这个场景应该用的不多，现在比较火的微服务，要么用dubbo，要么用spring cloud，用RabbitMQ做RPC比较少见，所以就先跳过了，有需要再补充。

其实网上RabbitMQ和Spring集成的教程有不少，我也大致看了看，大部分都是言简意赅，代码配置一贴，然后就可以用了，而我希望我的教程能多和大家一起探讨一些“为什么”。

# 2.Spring AMQP #

Spring AMQP中有两个单词，Spring都知道，那AMQP是什么？

中文意思是，高级消息队列协议，然后用蹩脚的英语猜一下，advance message queue protocol，差不多了，advance变成形容词高级的-advanced，queue变成queuing（排队论，学术一点），所以，AMQP就是Advanced Message Queuing Protocal。

AMQP 0-9-1 是RabbitMQ支持的协议之一，0-9-1是个版本号，正常情况下推荐使用，它所表现出的形式，就是前面几张介绍的内容

RabbitMQ还支持其他版本的协议，具体可以参考[这里][Link 1]

Spring AMQP的定义如下

> The Spring AMQP project applies core Spring concepts to the development of AMQP-based messaging solutions. It provides a "template" as a high-level abstraction for sending and receiving messages. It also provides support for Message-driven POJOs with a "listener container". These libraries facilitate management of AMQP resources while promoting the use of dependency injection and declarative configuration

意外的发现谷歌翻译的很通顺

Spring AMQP项目将核心Spring概念应用于基于AMQP的消息传递解决方案的开发。 它提供了一个“模板”作为发送和接收消息的高级抽象。 它还通过“侦听器容器”为消息驱动的POJO提供支持。 这些库促进AMQP资源的管理，同时促进使用依赖注入和声明性配置

[这里有很多资料][Link 2]

# 3.main方法集成 #

先不用Spring，只在Main方法中集成Spring AMQP，方便分析。

在pom文件中加入如下配置，不需要再引入RabbitMQ的包，Spring AMQP已经包含了4.0.x的客户端

>  Spring AMQP now uses the new 4.0.x version of the `amqp-client` library provided by the RabbitMQ team. This client has auto recovery configured by default

![复制代码][copycode.gif]

<dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>1.7.1.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.1</version>
    </dependency>

![复制代码][copycode.gif]

新建一个类，

![复制代码][copycode.gif]

1 package com.liyang.ticktock.rabbitmq;
     2 
     3 import org.springframework.amqp.core.BindingBuilder;
     4 import org.springframework.amqp.core.Queue;
     5 import org.springframework.amqp.core.TopicExchange;
     6 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
     7 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
     8 import org.springframework.amqp.rabbit.core.RabbitAdmin;
     9 import org.springframework.amqp.rabbit.core.RabbitTemplate;
    10 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    11 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
    12 
    13 
    14 public class App 
    15 {
    16     public static void main( String[] args ) throws InterruptedException
    17     {
    18         //获取一个连接工厂，用户默认是guest/guest（只能使用部署在本机的RabbitMQ）
    19         //是Spring实现的对com.rabbitmq.client.Connection的包装
    20         ConnectionFactory cf = new CachingConnectionFactory("localhost");
    21         
    22         //对AMQP 0-9-1的实现
    23         RabbitAdmin admin = new RabbitAdmin(cf);
    24         //声明一个队列
    25         Queue queue = new Queue("myQueue");
    26         admin.declareQueue(queue);
    27         //声明一个exchange
    28         TopicExchange exchange = new TopicExchange("myExchange");
    29         admin.declareExchange(exchange);
    30         //绑定队列到exchange，加上routingKey foo.*
    31         admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo.*"));
    32 
    33         //监听容器
    34         SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
    35         //监听者对象
    36         Object listener = new Object() {
    37             @SuppressWarnings("unused")
    38             public void handleMessage(String foo) {
    39                 System.out.println(foo);
    40             }
    41         };
    42         //通过这个适配器代理listener
    43         MessageListenerAdapter adapter = new MessageListenerAdapter(listener);
    44         //把适配器（listener）设置给Container
    45         container.setMessageListener(adapter);
    46         //设置该容器监听的队列名，可以传多个，public void setQueueNames(String... queueName) {
        
    47         container.setQueueNames("myQueue");
    48         //开始监听
    49         container.start();
    50 
    51         //发送模版，设置上连接工厂
    52         RabbitTemplate template = new RabbitTemplate(cf);
    53         //发送消息
    54         template.convertAndSend("myExchange", "foo.bar", "Hello, world!");
    55         
    56         Thread.sleep(1000);
    57         container.stop();
    58     }
    59 }

![复制代码][copycode.gif]

运行结果如下：

![1120662-20170410162604688-635309759.png][]

以上代码中

ConnectionFactory包装了所有物理连接信息，然后传递给RabbitAdmin创建了RabbitMQ支持协议的连接（AMQP 0-9-1）；

声明队列和交换中心后，通过BindingBuilder把队列的绑定关系声明到admin上；

创建一个消息处理类，用一个适配器（MessageListenerAdapter）包装它，并注册到监听容器中，启动监听；

最后通过连接信息创建一个Rabbit模板，调用发送方法。

这里引申出一个问题，为什么要用监听容器（SimpleMessageListenerContainer），我们点开它的outline，如下，可以看到，它是对监听这个动作的抽象，一个容器可以有多个Consumer，并且可以控制如超时时间等配置。

![1120662-20170410151540813-1881693360.png][]

可以看出，比起直接使用RabbitMQ客户端，以上代码已经简化了一部分，最明显的部分就是，不需要手动去关Channel、Connection了，对以上概念的介绍都穿插在前几章，这里不再赘述。

这些代码用Spring的方式注入将会更加简洁

# 4.传统Spring方式集成 #

首先，POM文件中要把Spring引进来

<dependency>
        <groupId>org.springframework</groupId>
             <artifactId>spring-context</artifactId>
        <version>4.3.7.RELEASE</version>
    </dependency>

然后增加applicatioContext.xml

![复制代码][copycode.gif]

<?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/beans  http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd
        http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd ">
    
        <context:annotation-config/>
        <context:component-scan base-package="com.liyang.ticktock.rabbitmq.listener"/>
    
        
        <rabbit:connection-factory id="connectionFactory"
            host="127.0.0.1" username="guest" password="guest" port="5672" />
    
        
        <rabbit:admin connection-factory="connectionFactory" />
    
        
        <rabbit:queue name="myQueue" />
    
        
        <rabbit:topic-exchange name="myExchange">
            <rabbit:bindings>
                <rabbit:binding queue="myQueue" pattern="foo.*" />
                
            </rabbit:bindings>
        </rabbit:topic-exchange>
    
        
        <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
            exchange="myExchange"/>
    
        
        <rabbit:listener-container connection-factory="connectionFactory">
            <rabbit:listener ref="myListener" method="listen" queue-names="myQueue" />
            
        </rabbit:listener-container>
    
    </beans>

![复制代码][copycode.gif]

这个配置文件中，我们使用了context的自动扫描装配，不用配bean了，还使用了rabbit的命名空间 xmlns:rabbit

这个东西可能有些童鞋不知道怎么配的，其实没什么神秘的，我们打开根路径看看里面有什么“http://www.springframework.org/schema/”

![1120662-20170410164342438-1485823035.png][]

和本地目录一样，就是个存东西的地方，rabbit命名空间，在这个路径后面加上rabbit对应的目录就行了

再进入rabbit目录，发现里面是一些不同版本的xsd文件，所以，在schemaLocation中，配上使用的具体版本的xsd就可以了

![1120662-20170410164443969-869160443.png][]

言归正传，上面的配置文件，把大量原来在代码中实现的东西挪到了xml中，一目了然，方便修改。注意注释加粗的三处，现在我们可以用Spring实现我们原来用原生RabbitMQ客户端实现的所有功能了

在发送消息之前，还需要写一个类用来处理消息，并且这个类要有linsten方法，和配置文件中声明的 method="listen" 对应

![复制代码][copycode.gif]

1 package com.liyang.ticktock.rabbitmq.listener;
     2 
     3 import org.slf4j.Logger;
     4 import org.slf4j.LoggerFactory;
     5 import org.springframework.stereotype.Component;
     6 
     7 @Component
     8 public class MyListener {
     9     Logger logger = LoggerFactory.getLogger(MyListener.class);
    10     
    11     public void listen(String message){
    12         logger.debug("received:"+message);
    13     }
    14 
    15 }

![复制代码][copycode.gif]

剩下在java中的代码就非常简洁了，监听在Spring环境启动后就自动开始了，我们只需要发消息就行，代码如下

直接一个main方法，省的还要捉一只汤姆猫

![复制代码][copycode.gif]

1 public static void main(String[] args) throws InterruptedException {
     2     //启动Spring环境
     3     AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
     4     //假装是Autowired的
     5     RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
     6     //设置routingKey
     7     template.setRoutingKey("foo.bar");
     8     //发送，exchange，routingKey什么的都配好了
     9     template.convertAndSend("Hello, world!");
    10         
    11     //关掉环境
    12     Thread.sleep(1000);
    13     ctx.destroy();
    14 }

![复制代码][copycode.gif]

运行结果如下：

![1120662-20170410165216422-1245466871.png][]

# 5.采用Spring Boot集成 #

不是很重要，先欠着，有空补上

# 6.结束语 #

最近实在太忙，这篇博客利用工作的碎片时间，写了有一个多星期，囧。

在写这篇系列教程的同时，我也获益良多，后面会继续为大家介绍一些RabbitMQ实用的高级特性，请多多支持

[Link 1]: https://www.rabbitmq.com/protocol.html
[Link 2]: http://docs.spring.io/spring-amqp/docs/1.7.1.RELEASE/reference/html/_reference.html#auto-recovery
[copycode.gif]: /images/20220527/576e40cf6c4d491f830a004c7f815ba7.png
[1120662-20170410162604688-635309759.png]: /images/20220527/ac9a299ca2f14504ad6a391efce0ee0d.png
[1120662-20170410151540813-1881693360.png]: /images/20220527/6d554e85cf414212b64b0cb4bbcfb7f9.png
[1120662-20170410164342438-1485823035.png]: /images/20220527/5e7f54bead9b4745acf9cf066dd53dd4.png
[1120662-20170410164443969-869160443.png]: /images/20220527/fc604b8ab50841258c2c465aee2f98ad.png
[1120662-20170410165216422-1245466871.png]: /images/20220527/e901ecca9ad1456d8483d3a2ed141cd8.png