Spring Boot集成RabbitMQ实现消息队列生产者与消费者_java springboot 配置rabbitmq消费者

目录

一、步骤概览

二、步骤说明

1.引入依赖

2.设置配置参数

①.RabbitMQ 服务器配置

②.消息模板（template）配置

③.消费端监听器（listener）配置

3.定义业务队列

①.定义交换机

②.定义队列

③.绑定队列和交换机

④.设置死信队列

4.生产端发送消息

①.处理流程

②.创建订单代码示例

③.异步通知发送结果

5.消费端消费消息

三、代码测试

1.测试代码

2.测试结果

①.生产端生成的100条消息

②.消费端消费日志

---

一、步骤概览

二、步骤说明

1.引入依赖

在 pom.xml 文件中引入 spring-boot-starter-amqp 依赖包

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.设置配置参数

在 application.yml 配置文件中设置 rabbitmq 配置参数。

spring:
 
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1 //你的RabbitMQ服务器地址
    port: 5672
    username: guest //用户名  
    password: guest //密码
    virtual-host: /
    template:
      retry:
        enabled: true
        max-attempts: 5
      publisher-returns: true
      publisher-confirm-type: correlated
 
    ## 消费端配置
    listener:
      simple:
        concurrency: 5
        ## manual:手动 ack（确认）
        acknowledge-mode: manual
        max-concurrency: 10
        prefetch: 1

这段 RabbitMQ 配置信息主要包含以下三部分：

①.RabbitMQ 服务器配置

• host：RabbitMQ 服务器的地址，这里设置为 127.0.0.1。

• port：RabbitMQ 服务器的端口，这里设置为 5672。

• username 和 password：连接 RabbitMQ 服务器的用户名和密码，这里设置为默认值 guest。

• virtual-host：虚拟主机的名称，用于逻辑隔离不同应用之间的消息队列，这里设置为根目录 /。

②.消息模板（template）配置

• retry.enabled：是否启用消息发送重试功能，这里设置为 true。

• retry.max-attempts：最大重试次数，这里设置为 5。

• publisher-returns：是否启用生产者消息发送失败返回功能，这里设置为 true。

• publisher-confirm-type：生产者消息确认方式，这里设置为 correlated。

③.消费端监听器（listener）配置

• simple.concurrency：消费者的并发消费数量，这里设置为 5。

• acknowledge-mode：消息确认模式，这里设置为手动 ack（确认）模式。

• max-concurrency：最大的并发消费数量，这里设置为 10。

• prefetch：每个消费者预取的消息数量，这里设置为 1。

3.定义业务队列

我们就以平时常见的商品购买为例，订单下完了，需要通知发货，我们就可以使用消息队列对其进行解耦。这边我们就定义订单队列。代码概览如下图所示

①.定义交换机

• RabbitQueueConfig#orderTopicExchange

@Bean
public TopicExchange orderTopicExchange() {
  return new TopicExchange(RabbitConstants.ORDER_EXCHANGE,true,false);
}

②.定义队列

• RabbitQueueConfig#orderQueue

@Bean
public Queue orderQueue() {
  //创建队列构造器并指定队列名称
  QueueBuilder queueBuilder = QueueBuilder.durable(RabbitConstants.ORDER_QUEUE);
  //如果队列持久化,这边不用设置队列过期时间
  //queueBuilder.ttl(orderQueueTTL)
  //设置死信队列的RouteKey
  queueBuilder.deadLetterRoutingKey(RabbitConstants.DEAD_ROUTE_KEY);
  //设置死信队列的Exchange
  queueBuilder.deadLetterExchange(RabbitConstants.DEAD_EXCHANGE);
  //创建队列
  return queueBuilder.build();
}

③.绑定队列和交换机

• RabbitQueueConfig#orderBinding

@Bean
public Binding orderBinding() {
  return BindingBuilder.bind(orderQueue())
    .to(orderTopicExchange())
    .with(RabbitConstants.ORDER_ROUTE_KEY);
}

④.设置死信队列

设置死信队列的目的是当消费者在消费消息时发生错误或者消息被拒绝，并且不重新入队（requeue=false）时，这些消息会被认为是消费失败的消息，并进入死信队列。 具体来说，消费失败的情况包括：

• 消费者抛出异常或者处理消息时发生错误。

• 消费者手动拒绝消息（basic.reject）并将 requeue 参数设置为 false。

• 消费者手动确认消息（basic.ack）之前，连接关闭导致消息未确认。

// 死信队列
@Bean
public Queue deadQueue() {
  return new Queue(RabbitConstants.DEAD_QUEUE, true);
}
// 死信队列使用的交换机
@Bean
public TopicExchange deadExchange() {
  return new TopicExchange(RabbitConstants.DEAD_EXCHANGE);
}
// 绑定交换机和死信队列
@Bean
public Binding deadBinding() {
  return BindingBuilder.bind(deadQueue())
    .to(deadExchange())
    .with(RabbitConstants.DEAD_ROUTE_KEY);
}

4.生产端发送消息

①.处理流程

②.创建订单代码示例

• OrderServiceImpl#createOrder

@Transactional(rollbackFor = Exception.class, isolation = Isolation.DEFAULT)
public void createOrder(Order order) {
 
    Date orderTime = new Date();
    int addrow = orderMapper.insert(order);
    if (addrow <= 0) {
        throw new RuntimeException("订单入库失败");
    }
    log.info("订单入库成功");
 
    // 插入消息记录表数据
    BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
 
    // 消息唯一ID
    brokerMessageLog.setMessageId(order.getMessageId());
 
    // 保存消息整体 转为JSON格式存储入库
    brokerMessageLog.setMessage(JSONUtil.toJsonStr(order));
 
    // 设置消息状态为0 表示发送中
    brokerMessageLog.setStatus(BrokerMsgStatusEnum.SENDING.getCode());
 
    // 设置消息未确认超时时间窗口为 一分钟
    brokerMessageLog.setNextRetry(DateUtil.offset(orderTime, DateField.MINUTE,RabbitConstants.ORDER_ACK_TIMEOUT));
    brokerMessageLog.setCreateTime(new Date());
    brokerMessageLog.setUpdateTime(new Date());
    addrow = messageLogMapper.insert(brokerMessageLog);
    if (addrow <= 0) {
        throw new RuntimeException("订单生成失败");
    }
    log.info("rabbitMq消息日志入库成功");
 
    // 发送消息
    CorrelationData correlationData = new CorrelationData(order.getMessageId());
    rabbitTemplate.convertAndSend(RabbitConstants.ORDER_EXCHANGE,
RabbitConstants.ORDER_ROUTE_KEY, JSONUtil.toJsonStr(order), correlationData);
}

③.异步通知发送结果

想要获取异步通知发送结果，订单接口需要实现 RabbitTemplate.ConfirmCallback 和 RabbitTemplate.ReturnsCallback 接口，分别重写 confirm 和 returnedMessage 方法，他们分别的作用是：

• RabbitTemplate.ConfirmCallback：主要用于处理生产者发送消息后的确认结果。当消息成功发送到 RabbitMQ 服务器时，会调用 ConfirmCallback 接口中的 confirm 方法，并传入该消息的相关信息（包括消息 ID、交换机、路由键等）。

/**
   * confirm机制只保证消息到达exchange,不保证消息路由到正确quene
   * 如果exchange错误,就会触发confirm机制
   *
   * @param correlationData 消息关联的数据
   * @param ack             消息是否正确到达exchange,true-到达;false-未到达
   * @param s               失败原因
   */
@Override
public void confirm(CorrelationData correlationData, boolean ack, String s) {
    log.info("correlationData:[{}],ack:[{}],cause:[{}]", correlationData, ack, s);
    String messageId = correlationData.getId();
    if (ack) {
        //如果confirm返回成功 则进行更新消息投递日志状态
        BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
        brokerMessageLog.setMessageId(messageId);
        brokerMessageLog.setUpdateTime(new Date());
        brokerMessageLog.setStatus(BrokerMsgStatusEnum.SUCCESS.getCode());
        messageLogMapper.updateById(brokerMessageLog);
        log.info("消息confirm成功,更新DB消息投递记录状态");
    } else {
        log.error("消息confirm失败,可尝试重试 或者补偿等手段");
    }
}

• RabbitTemplate.ReturnsCallback：用于处理生产者发送的消息无法被正确路由到队列时的返回结果。当消息无法被正确路由到队列时，RabbitMQ 会将该消息返回给生产者，并调用 ReturnsCallback 接口中的 returnedMessage 方法，并传入该消息的相关信息（包括消息 ID、交换机、路由键等）。通过实现 ReturnsCallback 接口，我们可以对无法路由的消息进行处理，例如记录日志、重新发送到其他队列等。

/**
   * Return 消息机制用于处理一个不可路由的消息,触发条件有以下两种情况:
   * 1. exchange不存在导致消息不可达
   * 2. routingKey路由不到导致消息不可达
   *
   * @param returnedMessage 返回信息
   */
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
log.error("路由信息不可达,message:[{}]", returnedMessage);
}

5.消费端消费消息

消费端只需要在接口上添加 @RabbitListener(queues = "order_queue_test") 就可以接收到队列消息了，其中 order_queue_test 表示消费的队列名称

• OrderReceiverService

@Slf4j
@Service
public class OrderReceiverService {
 
    @RabbitListener(queues = "order_queue_test")
    public void onOrderMessage(@Payload String body, 
                               @Headers Map<String,Object> headers, 
                               Channel channel) throws IOException {
        log.info("接受到订单信息:[{}]",body);
 
        /**
         * Delivery Tag 用来标识信道中投递的消息。
         * RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag，
         * 以便 Consumer 可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了。
         * RabbitMQ 保证在每个信道中，每条消息的 Delivery Tag 从 1 开始递增。
         */
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
 
        /**
         *  multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认
         *  如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认
         */
        boolean multiple = false;
 
        //手动ACK,确认一条消息已经被消费
        channel.basicAck(deliveryTag,multiple);
    }
}

三、代码测试

1.测试代码

• 模拟生成100条订单

@SpringBootTest
public class ApiTest {
    @Autowired
    private IOrderService orderService;
 
    @Test
    public void test_createOrder() throws Exception {
        for(int i= 1;i<=100;i++) {
            Order order = new Order();
            order.setMessageId(UUID.randomUUID().toString());
            order.setName("第" + i + "条订单");
            orderService.createOrder(order);
            TimeUnit.SECONDS.sleep(1);
        }
    }
}

2.测试结果

①.生产端生成的100条消息

②.消费端消费日志

从消费日志中我们可以看出，最多出现了6个消费线程进行消费，那是由于我们配置了最大消费线程数是10，rabbitmq 会动态调整消费者的数量进行消费，当消息消费完，它会恢复初始的消费者数量，我们定义的是5，mq 情况如图所示: