【RabbitMQ】SpringBoot整合RabbitMQ：工作队列 && 发布/订阅模式 && 路由模式 && 通配符模式

文章目录

• • 一、Work Queue（工作队列模式）
  • • 引入依赖
    • 添加配置
    • 编写生产者代码
    • 编写消费者代码
    • 运行结果
  • 二、Publish/Subscribe（发布/订阅模式）
  • • 编写生产者代码
    • 编写消费者代码
  • 消费者另一种写法
  • 三、Routing（路由模式）
  • • 编写生产者代码
    • 编写消费者代码
  • 四、Topics（通配符模式）
  • • 编写生产者代码
    • 编写消费者代码

Spring 官方：
Spring AMQP
RabbitMQ 官方：
RabbitMQ tutorial – “Hello World!” | RabbitMQ

一、Work Queue（工作队列模式）

步骤：（后面其它模式也是如此）

引入依赖

<!–Spring MVC相关依赖–>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>

<!–RabbitMQ相关依赖–>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

添加配置

# 配置RabbitMQ的基本信息
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: liren
password: 123123
virtual-host: lirendada

编写生产者代码

常量类：

public class Constants {
public static final String WORK_QUEUE = "work_queue";
}

然后在 config 包中声明队列：（注意包要导对~）

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
@Bean("workQueue")
public Queue WorkQueue() {
return QueueBuilder.durable(Constants.WORK_QUEUE).build();
}
}

最后在需要发送消息的地方调用 RabbitTemplate 发送消息：

@RequestMapping("/producer")
@RestController
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;

@RequestMapping("/work")
public String work(){
for (int i = 0; i < 10; i++) {
// 使用内置交换机发送消息, routingKey和队列名称保持一致
rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE, "hello spring amqp: work…");
}
return "发送成功";
}
}

编写消费者代码

定义监听类，用于消费队列中的消息：（注意包要导对~）

import com.liren.springbootrabbitmq.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class WorkListener {
@RabbitListener(queues = Constants.WORK_QUEUE)
public void workqueue1(Message message) {
System.out.println("workqueue1 [" + Constants.WORK_QUEUE + "]收到消息：" + message);
}

@RabbitListener(queues = Constants.WORK_QUEUE)
public void workqueue2(String message) {
System.out.println("workqueue2 [" + Constants.WORK_QUEUE + "]收到消息：" + message);
}
}

@RabbitListener 是 Spring 框架中用于监听 RabbitMQ 队列的注解，通过使用这个注解，可以定义一个方法，以便从 RabbitMQ 队列中接收消息。该注解支持多种参数类型，这些参数类型代表了从 RabbitMQ 接收到的消息和相关信息。

以下是一些常用的参数类型：

运行结果

运行程序，然后发起请求，会有三个队列接收消息，如下所示：

管理页面中可以看到三个消费者以及一个生产者通道：

二、Publish/Subscribe（发布/订阅模式）

RabbitMQ 交换机常见三种类型：fanout、direct、topic，不同类型有着不同的路由策略。

编写生产者代码

常量类：

// 发布订阅模式
public static final String FANOUT_QUEUE1 = "fanout.queue1";
public static final String FANOUT_QUEUE2 = "fanout.queue2";
public static final String FANOUT_EXCHANGE = "fanout.exchange";

然后在 config 包中声明队列：（注意包要导对~）

// 发布订阅模式
@Bean("publishConfirmQueue1")
public Queue publishConfirmQueue1() {
return QueueBuilder.durable(Constants.FANOUT_QUEUE1).build(); // 声明队列
}
@Bean("publishConfirmQueue2")
public Queue publishConfirmQueue2() {
return QueueBuilder.durable(Constants.FANOUT_QUEUE2).build(); // 声明队列
}

@Bean("fanoutExchange")
public FanoutExchange fanoutExchange() {
return ExchangeBuilder.fanoutExchange(Constants.FANOUT_EXCHANGE).build(); // 声明交换机
}

@Bean("fanoutBinding1")
public Binding fanoutBinding1(@Qualifier("publishConfirmQueue1") Queue queue,
@Qualifier("fanoutExchange") FanoutExchange exchange) {
return BindingBuilder.bind(queue).to(exchange); // 绑定交换机和队列
}
@Bean("fanoutBinding2")
public Binding fanoutBinding2(@Qualifier("publishConfirmQueue2") Queue queue,
@Qualifier("fanoutExchange") FanoutExchange exchange) {
return BindingBuilder.bind(queue).to(exchange); // 绑定交换机和队列
}

使用接口发送消息

@RequestMapping("/fanout")
public String fanout() {
rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE, "", "hello spring amqp: fanout…");
return "发送成功！";
}

编写消费者代码

@Component
public class FanoutListener {
@RabbitListener(queues = Constants.FANOUT_QUEUE1)
public void fanoutQueue1(String message) {
System.out.println("fanoutQueue1 [" + Constants.FANOUT_QUEUE1 + "]收到消息：" + message);
}

@RabbitListener(queues = Constants.FANOUT_QUEUE2)
public void fanoutQueue2(String message) {
System.out.println("fanoutQueue2 [" + Constants.FANOUT_QUEUE2 + "]收到消息：" + message);
}
}

消费者另一种写法

@RabbitListener 是一个功能强大的注解。这个注解里面可以配置 @QueueBinding、@Queue、@Exchange，直接通过这个组合注解一次性搞定多个交换机、绑定、路由、并且配置监听功能等

@Slf4j
@Component
public class UserRegisterListener {
@RabbitListener(
bindings = @QueueBinding(
value = @Queue(
value = Constants.USER_QUEUE_NANE, // 队列名
durable = "true" // 是否持久化
),
exchange = @Exchange(
value = Constants.USER_EXCHANGE_NAME, // 交换机名
type = ExchangeTypes.FANOUT // fanout 交换机
)
// fanout 不需要 routingKey
)
)
public void MailListenerQueue(Message message, Channel channel) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
// 处理用户注册消息
String body = new String(message.getBody());
log.info("用户注册消息处理成功，deliveryTag={}, message={}", deliveryTag, body);

// 发送邮件TODO

// 确认消息
channel.basicAck(deliveryTag, true);
}catch (Exception e) {
// 异常拒绝消息，进行重发
channel.basicNack(deliveryTag, true, true);
log.error("用户注册消息处理失败，拒绝消息，deliveryTag={}", deliveryTag, e);
}
}
}

启动时 Spring AMQP 会做的事情，顺序大致是：

三、Routing（路由模式）

RabbitMQ 交换机常见三种类型：fanout、direct、topic，不同类型有着不同的路由策略。

路由模式采用的是 RabbitMQ 中的 Direct 定向策略，生产者发送消息的时候，交换机需要根据消息中的 Routing Key 将消息发送给指定的队列，而不是发给每一个队列了！

此时，队列和交换机的绑定，不能是任意的绑定了，而是要指定一个 Binding Key。

只有队列绑定时的 Binding Key 和消息中的 Routing Key 完全一致，队列才会接收到消息。

编写生产者代码

常量类：

// 路由模式
public static final String DIRECT_EXCHANGE = "direct.exchange";
public static final String DIRECT_QUEUE1 = "direct.queue1";
public static final String DIRECT_QUEUE2 = "direct.queue2";

和发布订阅模式的区别是：交换机类型不同、绑定队列的 Binding Key 不同。

// 路由模式（direct模式）
@Bean("directQueue1")
public Queue directQueue1() {
return QueueBuilder.durable(Constants.DIRECT_QUEUE1).build(); // 声明队列
}
@Bean("directQueue2")
public Queue directQueue2() {
return QueueBuilder.durable(Constants.DIRECT_QUEUE2).build(); // 声明队列
}

@Bean("directExchange")
public DirectExchange directExchange() {
return ExchangeBuilder.directExchange(Constants.DIRECT_EXCHANGE).build(); // 声明交换机
}

// 队列1绑定orange
@Bean("directBinding1")
public Binding directBinding1(@Qualifier("directQueue1") Queue queue,
@Qualifier("directExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("orange"); // 绑定交换机和队列，以及bindingKey
}

// 队列2绑定green、black
@Bean("directBinding2")
public Binding directBinding2(@Qualifier("directQueue2") Queue queue,
@Qualifier("directExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("green"); // 绑定交换机和队列，以及bindingKey
}
@Bean("directBinding3")
public Binding directBinding3(@Qualifier("directQueue2") Queue queue,
@Qualifier("directExchange") DirectExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("black"); // 绑定交换机和队列，以及bindingKey
}

使用接口发送消息：

@RequestMapping("/direct/{routing_key}")
public String dirct(@PathVariable("routing_key") String routing_key) {
rabbitTemplate.convertAndSend(Constants.DIRECT_EXCHANGE, routing_key, "hello spring amqp: direct…" + routing_key);
return "发送成功！";
}

编写消费者代码

@Component
public class DirectListener {
@RabbitListener(queues = Constants.DIRECT_QUEUE1)
public void directQueue1(String message) {
System.out.println("directQueue1 [" + Constants.DIRECT_QUEUE1 + "]收到消息：" + message);
}

@RabbitListener(queues = Constants.DIRECT_QUEUE2)
public void directQueue2(String message) {
System.out.println("directQueue2 [" + Constants.DIRECT_QUEUE2 + "]收到消息：" + message);
}
}

分别请求三个不同的 routingkey，结果如下所示：

四、Topics（通配符模式）

Topics 和 Routing 模式的区别是：

匹配规则有如下要求：

编写生产者代码

常量类：

// 通配符模式
public static final String TOPIC_EXCHANGE = "topic.exchange";
public static final String TOPIC_QUEUE1 = "topic.queue1";
public static final String TOPIC_QUEUE2 = "topic.queue2";

生产者代码如下所示：

// 通配符模式（topics模式）
@Bean("topicQueue1")
public Queue topicQueue1() {
return QueueBuilder.durable(Constants.TOPIC_QUEUE1).build(); // 声明队列
}
@Bean("topicQueue2")
public Queue topicQueue2() {
return QueueBuilder.durable(Constants.TOPIC_QUEUE2).build(); // 声明队列
}

@Bean("topicExchange")
public TopicExchange topicExchange() {
return ExchangeBuilder.topicExchange(Constants.TOPIC_EXCHANGE).build(); // 声明交换机
}

// 队列1绑定error, 仅接收error信息
@Bean("topicBinding1")
public Binding topicBinding1(@Qualifier("topicQueue1") Queue queue,
@Qualifier("topicExchange") TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("*.error"); // 绑定交换机和队列，以及bindingKey
}
// 队列2绑定info, error: error,info信息都接收
@Bean("topicBinding2")
public Binding topicBinding2(@Qualifier("topicQueue2") Queue queue,
@Qualifier("topicExchange") TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("*.error"); // 绑定交换机和队列，以及bindingKey
}
@Bean("topicBinding3")
public Binding topicBinding3(@Qualifier("topicQueue2") Queue queue,
@Qualifier("topicExchange") TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("#.info"); // 绑定交换机和队列，以及bindingKey
}

使用接口发送消息：

@RequestMapping("/topics/{routing_key}")
public String topics(@PathVariable("routing_key") String routing_key) {
rabbitTemplate.convertAndSend(Constants.TOPIC_EXCHANGE, routing_key, "hello spring amqp: topics…" + routing_key);
return "发送成功！";
}

编写消费者代码

@Component
public class TopicListener {
@RabbitListener(queues = Constants.TOPIC_QUEUE1)
public void topicQueue1(String message) {
System.out.println("topicQueue1 [" + Constants.TOPIC_QUEUE1 + "]收到消息：" + message);
}

@RabbitListener(queues = Constants.TOPIC_QUEUE2)
public void topicQueue2(String message) {
System.out.println("topicQueue2 [" + Constants.TOPIC_QUEUE2 + "]收到消息：" + message);
}
}

分别请求两个不同的请求以及参数之后，运行结果如下：