Spring整合RabbitMQ

柔情只为你懂 2024-03-25 14:03 18阅读 0赞

## Spring整合RabbitMQ ##

### 1.生产者Spring-Producer ###

#### 1.1 创建生产者工程 ####

整合用到的maven坐标,主要有：`spring-context（Spring核心）`、`spring-test（测试）`、`junit`和`spring-rabbit（官方整合方案）`。

<dependencies>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.1.7.RELEASE</version>
        </dependency>
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.1.7.RELEASE</version>
            <scope>test</scope>
        </dependency>
        
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.1.8.RELEASE</version>
        </dependency>
    </dependencies>

#### 1.2 创建RabbitMQ配置文件 ####

这里我们创建的是一个RabbitMQ配置文件，名称为`rabbitmq.properties`。目的是方便将配置文件中的信息加载到Spring中，就无需在xml文件中修改了。

# RabbitMQ地址
    rabbitmq.host=192.168.52.128
    # RabbitMQ端口
    rabbitmq.port=5672
    # 用户名
    rabbitmq.username=admin
    # 密码
    rabbitmq.password=admin
    # 虚拟机
    rabbitmq.virtual-host=/test

#### 1.3 创建整合配置文件 ####

RabbitMQ和Spring使用xml文件进行整合，意味着大部分的配置类或模板都在xml文件中，我们默认的文件名为：`spring-rabbitmq-producer.xml`。

**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: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
           https://www.springframework.org/schema/context/spring-context.xsd
           http://www.springframework.org/schema/rabbit
           http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    </beans>

**（1）加载配置文件，并创建连接工程connectionFactory**

在这里我们需要注意的是加载配置文件的路径，`classpath:rabbitmq.properties`。

**（2）定义管理交换机、队列**

<rabbit:admin connection-factory="connectionFactory"/>

**（3）定义普通队列**

<rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/>

> 不绑定到交换机则绑定到默认交换机，默认交换机类型为direct，名字为：“”，路由键为队列的名称。

*  id：表示bean的id
 *  name：表示队列的名称
 *  auto-declare：如果没有是否自动创建
 *  auto-delete：是否自动删除队列（当最后一个消费者断开连接时）

**（4）定义广播类型交换机**

<rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/>
    
    
    <rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/>
    
    
    <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true">
        <rabbit:bindings>
            <rabbit:binding queue="spring_fanout_queue_1"/>
            <rabbit:binding queue="spring_fanout_queue_2"/>
        </rabbit:bindings>
    </rabbit:fanout-exchange>

**（5）定义定向类型交换机**

<rabbit:queue id="spring_direct_queue_1" name="spring_direct_queue_1" auto-declare="true"/>
    
    
    <rabbit:queue id="spring_direct_queue_2" name="spring_direct_queue_2" auto-declare="true"/>
    
    
    <rabbit:direct-exchange name="spring_direct_exchange">
        <rabbit:bindings>
            <rabbit:binding queue="spring_direct_queue_1" key="direct_key_1"/>
            <rabbit:binding queue="spring_direct_queue_2" key="direct_key_2"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

> 上述定义的是定向类型交换机，表示当routing key为direct\_key\_1的时候，会发送到spring\_direct\_queue\_1，当routing key为direct\_key\_2的时候，会发送到spring\_direct\_queue\_2。

**（6）定义通配符类型交换机**

<rabbit:queue id="spring_topic_queue_1" name="spring_topic_queue_1" auto-declare="true"/>
    
    <rabbit:queue id="spring_topic_queue_2" name="spring_topic_queue_2" auto-declare="true"/>
    
    <rabbit:queue id="spring_topic_queue_3" name="spring_topic_queue_3" auto-declare="true"/>
    
    <rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true">
        <rabbit:bindings>
            <rabbit:binding pattern="abc.*" queue="spring_topic_queue_1"/>
            <rabbit:binding pattern="abc.#" queue="spring_topic_queue_2"/>
            <rabbit:binding pattern="*.cd" queue="spring_topic_queue_3"/>
        </rabbit:bindings>
    </rabbit:topic-exchange>

> 上述代码表示：
> 
> （1）以`abc.`开头的routing key会转到`spring_topic_queue_1`，例如：`abc`、`abc.xx`、`abc.xx.xxx`。
> 
> （2）以`abc.`开头的且只有一个单词的会转到`spring_topic_queue_2`，例如：`abc.xx`。
> 
> （3）以".cd"结尾的的routing key会转到`spring_topic_queue_3`，例如：`abc.cd`,`xx.cd`。

*  pattern：表示的路由器的通配符
    
     *  \* 表示0或多个单词
     *  \# 表示一个单词

**（7）定义rabbitTemplate**

<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>

#### 1.4 代码测试 ####

@RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
    public class ProducerTest {
        
        /**
         * 注入RabbitTemplate
         */
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 测试默认的队列发送消息
         */
        @Test
        public void test() {
        
            rabbitTemplate.convertAndSend("spring_queue", "hello world");
        }
    
        /**
         * 测试广播模式发送消息
         */
        @Test
        public void testFanout() {
        
            rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "spring_fanout_exchange");
        }
    
        /**
         * 测试使用定向路由发送消息
         */
        public void testDirect() {
        
            // 将spring_direct_exchange_direct_key_1消息发送给路由key为direct_key_1的队列，即发送给spring_direct_queue_1
            rabbitTemplate.convertAndSend("spring_direct_exchange", "direct_key_1",
                    "spring_direct_exchange_direct_key_1");
            // 将spring_direct_exchange_direct_key_2消息发送给路由key为direct_key_2的队列，即发送给spring_direct_queue_2
            rabbitTemplate.convertAndSend("spring_direct_exchange", "direct_key_2",
                    "spring_direct_exchange_direct_key_2");
        }
    
        /**
         * 测试通配符模式
         */
        @Test
        public void testTopics() {
        
            rabbitTemplate.convertAndSend("spring_topic_exchange", "abc.cd", "spring_test_topics_abc.cd");
            rabbitTemplate.convertAndSend("spring_topic_exchange", "abc.x", "spring_test_topics_abc.x");
            rabbitTemplate.convertAndSend("spring_topic_exchange", "x.cd", "spring_test_topics_x.cd");
        }
    
    }

### 2.消费者Spring-Consumer ###

#### 2.1 创建消费者工程 ####

消费者Spring-Consumer使用的jar包和生产者一样，无需修改。

#### 1.2 创建RabbitMQ配置文件 ####

这里我们创建的是一个RabbitMQ配置文件，名称为`rabbitmq.properties`。目的是方便将配置文件中的信息加载到Spring中，就无需在xml文件中修改。

#### 1.3 创建整合配置文件 ####

RabbitMQ和Spring使用xml文件进行整合，意味着大部分的配置类或模板都在xml文件中，我们默认的文件名为：`spring-rabbitmq-producer.xml`。

**xml文件用到的默认约束**

**（1）加载配置文件，并创建连接工程connectionFactory**

在这里我们需要注意的是加载配置文件的路径，`classpath:rabbitmq.properties`。

<context:property-placeholder location="classpath:rabbitmq.properties"/>
    
    
    <rabbit:connection-factory id="connectionFactory"
                               host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"/>
    
    <bean id="springQueueListener" class="org.example.listener.SpringQueueListener"/>
    
    <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true">
    	<rabbit:listener ref="springQueueListener" queue-names="spring_queue"/>
    </rabbit:listener-container>

**（2）创建监听器Listener负责监听对应的队列消息**

我们创建一个默认普通的监听器，用来监听`spring_queue`队列中的消息。

监听器默认要实现`MessageListener`接口，并实现`onMessage`方法。

public class SpringQueueListener implements MessageListener {
        
        @Override
        public void onMessage(Message message) {
        
            System.out.println("SpringQueueListener:" + new String(message.getBody()));
        }
    }

**（3）单元测试**

@RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
    public class ConsumerTest {
        
    
        @Test
        public void test() {
        
            while (true) {
        
            }
        }
    }

> 控制台：SpringQueueListener:hello world