Spring整合RabbitMQ简介

怼烎@ 2022-07-03 23:53 297阅读 0赞

# 预备知识 #

RabbitMQ是AMQP(Advanced Message Queuing Protoco)高级消息队列协议的具体实现，在学习RabbitMQ之前，有必要先熟悉下AMQP。AMQP属于应用层协议，它有producer、exchange、binding、queue、consumer五种角色，producer将消息发送到exchange，bingding决定exchange的消息该发往哪个queue，consumer直接从queue中读取消息。exchange是AMQP协议的重要组件，消息发送到exchange，exchange将消息路由到零或多个消息队列中，路由算法由exchange类型决定。关于exchange类型可以参考网上的[这篇博客][Link 1]。AMQP区别于JMS(Java Message Service),具有天然的跨平台、跨语言及更详细的消息模型。JMS其实就是一个java消息服务API，依赖于java语言，实现如Apache ActiveMQ等。

# spring-rabbit #

## Maven依赖 ##

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

## Java方式实现 ##

spring-rabbit主要API有：

*  MessageListenerContainer：监听容器，为消息入队提供异步处理
 *  RabbitTemplate ：用来发送和接收消息
 *  RabbitAdmin ：声明队列、交换器（Exchange）、绑定（Binding）

代码清单：

package amqp;
    
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
    
    public class Main {
        public static void main(String[] args) throws InterruptedException {
            ConnectionFactory cf = new CachingConnectionFactory("127.0.0.1",5672);
            
            RabbitAdmin admin = new RabbitAdmin(cf);
            //创建队列
            Queue queue = new Queue("myQueue");
            admin.declareQueue(queue);
            //创建topic类型的交换机
            TopicExchange exchange = new TopicExchange("myExchange");
            admin.declareExchange(exchange);
            //交换机和队列绑定，路由规则为匹配"foo."开头的路由键
            admin.declareBinding(
                    BindingBuilder.bind(queue).to(exchange).with("foo.*"));
    
            //设置监听
            SimpleMessageListenerContainer container =
                    new SimpleMessageListenerContainer(cf);
            Object listener = new Object() {
                public void handleMessage(String foo) {
                    System.out.println(" [x] Received '" + foo + "'");
                }
            };
            MessageListenerAdapter adapter = new MessageListenerAdapter(listener);
            container.setMessageListener(adapter);
            container.setQueueNames("myQueue");
            container.start();
    
            //发送消息
            RabbitTemplate template = new RabbitTemplate(cf);
            template.convertAndSend("myExchange", "foo.bar", "Hello, world!");
            Thread.sleep(1000);
            container.stop();
        }
    }

## Spring方式实现 ##

*  **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: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/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
          
        
        <context:component-scan base-package="amqp" />
        
        
        <rabbit:connection-factory id="connectionFactory" host="127.0.0.1" port="5672"
                                   username="guest" password="guest"  virtual-host="/" requested-heartbeat="60" />
    
        
        <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
                         exchange="myExchange" routing-key="foo.bar"/>
    
        
        <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:listener-container connection-factory="connectionFactory">
            <rabbit:listener ref="foo" method="listen" queue-names="myQueue" />
        </rabbit:listener-container>
    
    </beans>

*  **监听处理程序**

package amqp;
    
    @Component
    public class Foo {
        public void listen(String message) {
            System.out.println(" [x] Received '" + message + "'");
        }
    }

实际项目中通常实现MessageListener接口来监听消息

package amqp;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class FooMessageListener implements MessageListener {
    
        @Override
        public void onMessage(Message message) {
            String messageBody = new String(message.getBody());
            System.out.println(" [x] Received '" + messageBody + "'");
        }
    }

配置文件做相应修改，此时不需要再指定处理方法

<rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener ref="fooMessageListener" queue-names="myQueue" />
    </rabbit:listener-container>

*  **发送消息**

package amqp;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.support.AbstractApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    
    public class SpringMain {
        public static void main(final String... args) throws Exception {
    
            AbstractApplicationContext ctx =
                    new ClassPathXmlApplicationContext("context.xml");
            RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
            template.convertAndSend("Hello, world!");
            Thread.sleep(1000);
            ctx.destroy();
        }
    }

[Link 1]: http://hwcrazy.com/34195c9068c811e38a44000d601c5586/be62fc2668c811e3adba000d601c5586/