教程说明
- 本系列教程目录大纲：《RabbitMQ系列教程-目录大纲》
- 本系列教程配套代码：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>
<!--spring核心依赖-->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>5.1.7.RELEASE</version>
</dependency>
<!--spring整合rabbitmq依赖-->
<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>
<!--spring整合测试单元-->
<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"/>
<!-- 定义rabbitmq connectionFactory -->
<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"/>
<!--
定义一个队列
name: 队列的名称
durable: 是否持久化(默认true)
auto-delete: 是否自动删除(默认false)
exclusive: 是否独占(默认false)
-->
<rabbit:queue id="spring_work_queue" name="spring_work_queue" auto-delete="false"
exclusive="false" durable="true"/>
<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<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....");
}
}
两个方法底层都是采用默认的交换机（空字符串
""），效果一样
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"/>
<!-- 定义rabbitmq connectionFactory -->
<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扩展：
定义一个交换机和两个队列，并且把队列绑定到交换机
<!--
定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机
默认交换机类型为direct，名字为: ""，路由键为队列的名称
-->
<rabbit:queue id="spring_fanout_queue1" name="spring_fanout_queue1"/>
<rabbit:queue id="spring_fanout_queue2" name="spring_fanout_queue2"/>
<!-- fanout工作模式的交换机 -->
<rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange">
<!--绑定queue-->
<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中注册两个监听器：
<bean id="springFanoutListener1" class="com.lscl.rabbitmq.SpringFanoutListener1"/>
<bean id="springFanoutListener2" class="com.lscl.rabbitmq.SpringFanoutListener2"/>
<!-- 绑定到监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="springFanoutListener1" queue-names="spring_fanout_queue1"/>
<rabbit:listener ref="springFanoutListener2" queue-names="spring_fanout_queue2"/>
</rabbit:listener-container>
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" />
<!-- routing工作模式 将队列绑定到交换机 -->
<rabbit:direct-exchange id="spring_direct_exchange" name="spring_direct_exchange" >
<rabbit:bindings>
<!--绑定queue指定routing key-->
<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中注册监听器：
<bean id="directListener1" class="com.lscl.rabbitmq.SpringDirectListener1"/>
<bean id="directListener2" class="com.lscl.rabbitmq.SpringDirectListener2"/>
<!-- 绑定到监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="directListener1" queue-names="spring_direct_queue1"/>
<rabbit:listener ref="directListener2" queue-names="spring_direct_queue2"/>
</rabbit:listener-container>
5.3.3 Topics模式
5.3.3.1 生产者
在spring.xml中定义交换机以及队列：
<!-- 定义4个队列 -->
<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" />
<!-- topic工作模式,将队列绑定到交换机 -->
<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中注册监听器：
<bean id="topicListener1" class="com.lscl.rabbitmq.SpringTopicListener1"/>
<bean id="topicListener2" class="com.lscl.rabbitmq.SpringTopicListener2"/>
<bean id="topicListener3" class="com.lscl.rabbitmq.SpringTopicListener3"/>
<bean id="topicListener4" class="com.lscl.rabbitmq.SpringTopicListener4"/>
<!-- 绑定到监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="topicListener1" queue-names="spring_topic_queue1"/>
<rabbit:listener ref="topicListener2" queue-names="spring_topic_queue2"/>
<rabbit:listener ref="topicListener3" queue-names="spring_topic_queue3"/>
<rabbit:listener ref="topicListener4" queue-names="spring_topic_queue4"/>
</rabbit:listener-container>
5.3.4 Header类型交换机
5.3.4.1 生产者
在spring.xml中编写：
<rabbit:queue id="spring_header_queue" name="spring_header_queue"/>
<!-- header模式,将队列绑定到交换机 -->
<rabbit:headers-exchange id="spring_header_exchange" name="spring_header_exchange">
<rabbit:bindings>
<rabbit:binding queue="spring_header_queue" key="key1" value="147"/>
<!-- <rabbit:binding queue="spring_header_queue" key="key2" value="258"/>-->
<!-- <rabbit:binding queue="spring_header_queue" key="key3" value="369"/>-->
</rabbit:bindings>
</rabbit:headers-exchange>
<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<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中注册监听器：
<bean id="headerListener" class="com.lscl.rabbitmq.SpringHeaderListener"/>
<!-- 绑定到监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory">
<!-- 引用监听器并指定监听的队列,监听多个队列以逗号分隔 -->
<rabbit:listener ref="headerListener" queue-names="spring_header_queue"/>
</rabbit:listener-container>
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/131799.html
