Docker版RabbitMQ安装延迟队列插件及延迟队列项目应用实战

Posted 2023-01-13 Galen-gao

前言

在项目中经常有延迟业务处理的背景，此时可以借助于Rabbitmq的延迟队列进行实现，但Rabbitmq本身并不支持延迟队列，但可以通过安装插件的方式实现延迟队列

环境准备

1. 首先确认目前项目使用的Rabbitmq的版本，这里博主的版本是3.9.15的。
   
2. 访问 Rabbitmq的github网址,检索 delay 找到插件rabbitmq-delayed-message-exchange ，如下图所示：
   
3. 找到延迟队列插件相应的版本并进行下载。
   
   博主的Rabbitmq是3.9版本的，所以这里选择3.9版本即可如下图所示：
   
   下载 rabbitmq_delayed_message_exchange-3.9.0.ez
   

安装延迟队列

上面我们已经成功下载了延迟队列插件，接下来我们需要将此插件上传到Linux系统服务器上，然后再将插件拷贝到Rabbitmq的容器中。

1. 查看Rabbitmq容器进程
   
2. 将Rabbitmq延迟队列插件拷贝到容器中

# c94f1e7dbfc0 代表容器Id，通过上面1的步骤即可查看
docker cp rabbitmq_delayed_message_exchange-3.9.0.ez c94f1e7dbfc0:/opt/rabbitmq/plugins

3. 进入容器并安装延迟队列插件

docker exec -it c94f1e7dbfc0 /bin/bash

查看 /opt/rabbitmq/plugins，是否已经有了延迟队列插件。如下图所示：

安装插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

4. 安装消息管理插件

 rabbitmq-plugins enable rabbitmq_shovel rabbitmq_shovel_management

4. 查看已经安装的Rabbitmq插件

5. 控制台查看验证延迟队列类型如下图：

延迟队列实战

SpringBoot项目中引入Rabbitmq的依赖项

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

新增一个交换机、一个队列然后绑定队列与交换机如下图所示：

@Configuration
public class RabbitmqConfig 
    @Bean(name = "updateAmount")
    public TopicExchange lazyExchange() 
        Map<String, Object> pros = new HashMap<>();
        //设置交换机支持延迟消息推送
        pros.put("x-delayed-type", "direct");
        TopicExchange exchange = new TopicExchange("updateAmount", true, false,pros);
        // 这一行是重点，指定交换机类型
        exchange.setDelayed(true);
        return exchange;
    

    @Bean
    public Queue lazyQueue()
        return new Queue("LAZY_QUEUE", true);
    

    @Bean
    public Binding lazyBinding()
        return BindingBuilder.bind(lazyQueue()).to(lazyExchange()).with("");
    

新增发送延迟队列的工具类代码逻辑：

public class RabbitmqUtils 


    /**
     * 发送延迟消息
     * @param rabbitTemplate rabbitTemplate
     * @param millisecond 延迟毫秒
     * @param messageContent 发送字符串
     * @param busiId 业务主键id
     */
    public static void sendDelayMessage(RabbitTemplate rabbitTemplate, Integer millisecond, Object messageContent, Long busiId) 
        CorrelationData correlationData = new CorrelationData(busiId.toString() + System.currentTimeMillis());
        rabbitTemplate.convertAndSend("updateAmount", "", messageContent,
                message -> 
                    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    message.getMessageProperties().setDelay(millisecond);
                    return message;
                , correlationData);
    

新增一个监听器代码消费延迟队列的消息：

@Component
@Slf4j
@RequiredArgsConstructor
@RabbitListener(queues = "LAZY_QUEUE")
public class RabbitmqListener 

    private final ObjectMapper objectMapper;


    @RabbitHandler
    public void onCustomBookingMessage(@Payload String message, Channel channel, @Headers Map<String, Object> headers) 
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try 
            Map map = objectMapper.readValue(message, Map.class);
            log.info("接收消息成功，消息内容："+map);
            ack(channel,deliveryTag,"");
         catch (JsonProcessingException e) 
            nack(channel,deliveryTag,"");
        
    


    /**
     * 确认消息成功
     */
    public void ack(Channel channel, long deliveryTag, String info) 
        try 
            channel.basicAck(deliveryTag, false);
         catch (IOException e) 
            log.error(info + ":消息应答出错", e);
        
    

    /**
     * 确认消息失败
     */
    public void nack(Channel channel, long deliveryTag, String info) 
        try 
            channel.basicNack(deliveryTag, false, true);
         catch (IOException e) 
            log.error(info + ":消息拒绝出错", e);
        
    

新增一个生产者发送消息的代码：

@RequestMapping("/rabbit")
@RequiredArgsConstructor
@RestController
@Slf4j
public class RabbitmqController 
    private final RabbitTemplate rabbitTemplate;

    private final ObjectMapper objectMapper;


    @GetMapping("/sendMessage")
    public void sendDeLayMessage() throws JsonProcessingException 
        Map<String,Object> map = new HashMap<>();
        map.put("username","zhangsan");
        map.put("tranferMoney",new BigDecimal("1200.23"));
        map.put("time", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        // 发送消息
        RabbitmqUtils.sendDelayMessage(rabbitTemplate,7000,objectMapper.writeValueAsString(map),System.currentTimeMillis());
        log.info("发送延迟队列消息成功：消息内容："+map);
    ;

在浏览器访问：http://127.0.0.1:8080/rabbit/sendMessage 后可以看到控制台输出如下日志：

因为我们设定延迟时间为7s，可以清晰看到日志打印的时间刚好相聚了7s，证明我们延迟队列发送消息成功。