《RabbitMQ系列教程-第五章-Spring整合RabbitMQ》

古城微笑少年丶 2023-01-02 03:26 189阅读 0赞

# 教程说明 #

*  本系列教程目录大纲：[《RabbitMQ系列教程-目录大纲》][RabbitMQ_-]

--------------------

*  本系列教程配套代码：[https://gitee.com/Horizon1024/rabbitmt.git][https_gitee.com_Horizon1024_rabbitmt.git]（码云地址）

--------------------

# 第五章 Spring整合RabbitMQ #

## 5.1 搭建消息生产者 ##

### pom.xml： ###

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.lscl</groupId>
        <artifactId>03_spring_rabbitmq_producer</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <dependencies>
            
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>5.1.7.RELEASE</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit</artifactId>
                <version>2.1.8.RELEASE</version>
            </dependency>
    
            
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-test</artifactId>
                <version>5.1.7.RELEASE</version>
            </dependency>
        </dependencies>
    </project>

### rabbitmq.properties： ###

rabbitmq.host=192.168.40.139
    rabbitmq.port=5672
    rabbitmq.username=lscl
    rabbitmq.password=admin
    rabbitmq.virtual-host=/lscl

### 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: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">
        
        <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}"/>
        
        <rabbit:admin connection-factory="connectionFactory"/>
    
        
        <rabbit:queue id="spring_work_queue" name="spring_work_queue" auto-delete="false" exclusive="false" durable="true"/>
    
        
        <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
    </beans>

### 消息生产者测试类： ###

package com.lscl.rabbitmq;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @ContextConfiguration(locations = "classpath:spring.xml")
    public class ProducerTest { 
    
        //1.注入 RabbitTemplate
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /** * work模式 */
        @Test
        public void testHelloWorld(){ 
            //2.发送消息(采用默认的交换机"")
            rabbitTemplate.convertAndSend("spring_work_queue","spring work....");
    		
    		// 也是采用默认的交换机""
    		// rabbitTemplate.convertAndSend("","spring_work_queue","spring work....");
        }
    
    }

![在这里插入图片描述][20210104170055341.png_pic_center]

> 两个方法底层都是采用默认的交换机（空字符串`""`），效果一样

## 5.2 搭建消息消费者 ##

### pom.xml： ###

和生产者一致

### rabbitmq.properties： ###

和生产者一致

### 监听器类： ###

需要实现MessageListener接口，重写onMessage方法

package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringWorkListener implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("work:" + new String(message.getBody()));
        }
    }

### 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: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">
        
        <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="springWorkListener" class="com.lscl.rabbitmq.SpringWorkListener"/>
    
        
        <rabbit:listener-container connection-factory="connectionFactory">
            
            <rabbit:listener ref="springWorkListener" queue-names="spring_work_queue"/>
    
        </rabbit:listener-container>
    </beans>

### 消息消费者测试类： ###

package com.lscl.rabbitmq;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @ContextConfiguration(locations = "classpath:spring.xml")
    public class ConsumerTest { 
    
        @Test
        public void test1(){ 
           
        }
    }

## 5.3 测试其他类型队列 ##

### 5.3.1 Pub/Sub模式 ###

**强调fanout类型的交换机，也叫分列模式**

#### 5.3.1.1 生产者 ####

在spring.xml扩展：

定义一个交换机和两个队列，并且把队列绑定到交换机

<rabbit:queue id="spring_fanout_queue1" name="spring_fanout_queue1"/>
    <rabbit:queue id="spring_fanout_queue2" name="spring_fanout_queue2"/>
    
    
    <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange">
        
        <rabbit:bindings>
            <rabbit:binding queue="spring_fanout_queue1"/>
            <rabbit:binding queue="spring_fanout_queue2"/>
        </rabbit:bindings>
    </rabbit:fanout-exchange>

编写测试类：

/** * Pub/sub模式(fanout模式) */
    @Test
    public void testFanout(){ 
        //2.发送消息
    
        rabbitTemplate.convertAndSend("spring_fanout_exchange","","spring fanout....");
    }

#### 5.3.1.2 消费者 ####

编写两个监听器：

package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringFanoutListener1 implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("fanout1:" + new String(message.getBody()));
        }
    }
    ------------------------------------------------------------------
    package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringFanoutListener2 implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("fanout2:" + new String(message.getBody()));
        }
    }

在spring.xml中注册两个监听器：

### 5.3.2 Routing模式 ###

**强调direct类型的交换机，也叫直连模式；**

#### 5.3.2.1 生产者 ####

扩展spring.xml：

<rabbit:queue id="spring_direct_queue1" name="spring_direct_queue1" />
    <rabbit:queue id="spring_direct_queue2" name="spring_direct_queue2" />
    
    
    <rabbit:direct-exchange id="spring_direct_exchange" name="spring_direct_exchange" >
        <rabbit:bindings>
            
            <rabbit:binding queue="spring_direct_queue1" key="red"/>
            <rabbit:binding queue="spring_direct_queue2" key="green"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

编写测试类：

/** * routing模式(direct模式) */
    @Test
    public void testRouting(){ 
        //2.发送消息
    
        rabbitTemplate.convertAndSend("spring_direct_exchange","red","spring routing red....");
        rabbitTemplate.convertAndSend("spring_direct_exchange","green","spring routing green....");
    }

#### 5.3.2.2 消费者 ####

编写两个监听器：

package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringDirectListener1 implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("direct1:" + new String(message.getBody()));
        }
    }
    ---------------------------------------------------------------------------------------
    package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringDirectListener2 implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("direct2:" + new String(message.getBody()));
        }
    }

在spring.xml中注册监听器：

### 5.3.3 Topics模式 ###

#### 5.3.3.1 生产者 ####

在spring.xml中定义交换机以及队列：

<rabbit:queue id="spring_topic_queue1" name="spring_topic_queue1" />
    <rabbit:queue id="spring_topic_queue2" name="spring_topic_queue2" />
    <rabbit:queue id="spring_topic_queue3" name="spring_topic_queue3" />
    <rabbit:queue id="spring_topic_queue4" name="spring_topic_queue4" />
    
    
    <rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" >
        <rabbit:bindings>
            <rabbit:binding pattern="red.#.green" queue="spring_topic_queue1"/>
            <rabbit:binding pattern="red.green.*" queue="spring_topic_queue2"/>
            <rabbit:binding pattern="#.green" queue="spring_topic_queue3"/>
            <rabbit:binding pattern="*.green.#" queue="spring_topic_queue4"/>
        </rabbit:bindings>
    </rabbit:topic-exchange>

编写测试方法：

/** * topic模式 */
    @Test
    public void testTopics() { 
        //2.发送消息
    
        // 命中: q1、q2、q3、q4
        rabbitTemplate.convertAndSend("spring_topic_exchange", "red.green.green", "spring topic....");
    
        // 命中: q3、q4
    // rabbitTemplate.convertAndSend("spring_topic_exchange","green.green","spring topic....");
    
        // 命中: 没有一个命中
    // rabbitTemplate.convertAndSend("spring_topic_exchange","green.red.blue","spring topic....");
    
        // 命中: q3、q4
    // rabbitTemplate.convertAndSend("spring_topic_exchange","green.green.green","spring topic....");
    
        // 命中: q3
    // rabbitTemplate.convertAndSend("spring_topic_exchange","green.red.green","spring topic....");
    }

#### 5.3.3.2 消费者 ####

编写4个监听器：

package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringTopicListener1 implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("topic1:" + new String(message.getBody()));
        }
    }

> 其余三个代码一模一样

在spring.xml中注册监听器：

### 5.3.4 Header类型交换机 ###

#### 5.3.4.1 生产者 ####

在spring.xml中编写：

<rabbit:queue id="spring_header_queue" name="spring_header_queue"/>
    
    
    <rabbit:headers-exchange id="spring_header_exchange" name="spring_header_exchange">
        <rabbit:bindings>
            <rabbit:binding queue="spring_header_queue" key="key1" value="147"/>
    
    
        </rabbit:bindings>
    </rabbit:headers-exchange>
    
    
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
    
    <bean id="rabbitMessagingTemplate" class="org.springframework.amqp.rabbit.core.RabbitMessagingTemplate">
        <property name="rabbitTemplate" ref="rabbitTemplate" />
    </bean>

发送消息：

@Autowired
    private RabbitMessagingTemplate rabbitMessagingTemplate;
    
    /** * header模式 * * @throws Exception */
    @Test
    public void tesHeader() throws Exception { 
    
        // 准备header参数
        Map<String, Object> headers = new HashMap<>();
    // headers.put("key1", "147");
        headers.put("key2", "258");
    // headers.put("key3", "369");
    
        // 使用的是rabbitMessagingTemplate 而不是 rabbitTemplate
        rabbitMessagingTemplate.convertAndSend("spring_header_exchange", "", "boot header....", headers);
    }

#### 5.3.4.2 消费者： ####

编写spring\_header\_queue队列的监听器：

package com.lscl.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class SpringHeaderListener implements MessageListener { 
        @Override
        public void onMessage(Message message) { 
            //打印消息
            System.out.println("header:" + new String(message.getBody()));
        }
    }

在spring.xml中注册监听器：

[RabbitMQ_-]: https://blog.csdn.net/Bb15070047748/article/details/112194912
[https_gitee.com_Horizon1024_rabbitmt.git]: https://gitee.com/Horizon1024/rabbitmt.git
[20210104170055341.png_pic_center]: /images/20221120/a838250a85a34eb2b09b9e4984136a2e.png