【SpringBoot】43、SpringBoot中整合RabbitMQ实现延时队列(延时插件篇)「建议收藏」

【SpringBoot】43、SpringBoot中整合RabbitMQ实现延时队列(延时插件篇)「建议收藏」一、介绍1、什么是延时队列?延时队列即就是放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费2、适用场景(1)商城订单超时未支付,取消订单(2)使用权限到期前十分钟提醒用户(3)收益项目,投入后一段时间后产生收益二、实现方式从以上场景中,我们可以看出,延时队列的主要功能就是在指定的时间之后做指定的事情,那么,我们思考有哪些工具我们可以使用?1、Redis监听过期Key可以参考我的博客【SpringBoot】三十五、SpringBoot整合Redis监听Ke

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

死信队列实现篇,参考文章:【SpringBoot】60、SpringBoot中整合RabbitMQ实现延时队列(死信队列篇)

一、介绍

  • 1、什么是延时队列?
    延时队列即就是放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费
  • 2、适用场景
    (1)商城订单超时未支付,取消订单
    (2)使用权限到期前十分钟提醒用户
    (3)收益项目,投入后一段时间后产生收益

二、实现方式

从以上场景中,我们可以看出,延时队列的主要功能就是在指定的时间之后做指定的事情,那么,我们思考有哪些工具我们可以使用?

  • 1、Redis 监听过期 Key

可以参考我的博客【SpringBoot】三十五、SpringBoot整合Redis监听Key过期事件

https://lizhou.blog.csdn.net/article/details/109238083
  • 2、RabbitMQ等实现延时队列

这也是本片文章中要讲的知识点,使用 RabbitMQ 实现延时队列有两种方式

(1)利用两个特性: Time To Live(TTL)、Dead Letter Exchanges(DLX)
(2)利用 RabbitMQ 中的插件 x-delay-message

本文主要讲解第二种方式,使用插件的方式

三、下载插件

RabbitMQ 实现了一个插件 x-delay-message 来实现延时队列,我们可以从 这里 下载到它

https://www.rabbitmq.com/community-plugins.html

选择 rabbitmq_delayed_message_exchange 插件,如图所示
下载插件
下载插件
选择 .ez 格式的文件下载,下载后放置 RabbitMQ 的安装目录下的 plugins 目录下,如我的路径为

D:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.16\plugins

执行命令

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

安装插件完成

四、在SpringBoot整合RabbitMQ

1、引入 RabbitMQ 依赖

<!-- rabbitmq消息队列 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、配置 RabbitMQ 信息

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        # 手动ACK 不开启自动ACK模式,目的是防止报错后未正确处理消息丢失 默认 为 none
        acknowledge-mode: manual

3、RabbitMQ 常量类

package com.asurplus.common.rabbitmq;

/**
 * rabbit常量类
 *
 * @Author Lizhou
 */
public final class RabbitConst {

    /**
     * 交换机
     */
    public static final String DELAY_EXCHANGE = "delay_exchange";

    /**
     * 队列
     */
    public static final String DELAY_QUEUE = "delay_queue";

    /**
     * 路由
     */
    public static final String DELAY_KEY = "delay_key";

}

4、RabbitMQ 配置类

package com.asurplus.common.rabbitmq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * rabbitmq配置类
 *
 * @Author Lizhou
 */
@Configuration
public class RabbitConfig {

    /**
     * 延时队列交换机
     *
     * @return
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(RabbitConst.DELAY_EXCHANGE, "x-delayed-message", true, false, args);
    }

    /**
     * 延时队列
     *
     * @return
     */
    @Bean
    public Queue delayQueue() {
        return new Queue(RabbitConst.DELAY_QUEUE, true);
    }

    /**
     * 给延时队列绑定交换机
     *
     * @return
     */
    @Bean
    public Binding delayBinding(Queue delayQueue, CustomExchange delayExchange) {
        return BindingBuilder.bind(delayQueue).to(delayExchange).with(RabbitConst.DELAY_KEY).noargs();
    }
}

5、RabbitMQ 生产者

package com.asurplus.common.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * rabbitMq生产者
 *
 * @Author Lizhou
 */
@Component
@Slf4j
public class RabbitProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     *
     * @param object      发送对象
     * @param millisecond 延时(毫秒)
     */
    public void sendDelayMessage(Object object, long millisecond) {
        this.rabbitTemplate.convertAndSend(
                RabbitConst.DELAY_EXCHANGE,
                RabbitConst.DELAY_KEY,
                object.toString(),
                message -> {
                    message.getMessageProperties().setHeader("x-delay", millisecond);
                    return message;
                }
        );
    }
}

6、RabbitMQ 消费者

package com.asurplus.common.rabbitmq;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * activeMq消费者
 *
 * @Author Lizhou
 */
@Component
@Slf4j
public class RabbitConsumer {

    /**
     * 接收消息
     *
     * @param object 监听的内容
     */
    @RabbitListener(queues = RabbitConst.DELAY_QUEUE)
    public void cfgUserReceiveDealy(Object object, Message message, Channel channel) throws IOException {
		// 通知 MQ 消息已被接收,可以ACK(从队列中删除)了
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        try {
            log.info("接受消息:{}", object.toString());
        } catch (Exception e) {
            log.error(e.getMessage());
            /**
             * basicRecover方法是进行补发操作,
             * 其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer(集群)接收到,
             * 设置为false是只补发给当前的consumer
             */
            channel.basicRecover(false);
        }
    }
}

五、测试

package com.asurplus;

import com.asurplus.common.rabbitmq.RabbitProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
@RestController
public class RabbitmqApplication {

    @Autowired
    private RabbitProducer product;

    @GetMapping("init")
    public void init() {
        String message1 = "这是第一条消息";
        String message2 = "这是第二条消息";
        product.sendDelayMessage(message1, 5000);
        product.sendDelayMessage(message2, 10000);
    }

    public static void main(String[] args) {
        SpringApplication.run(RabbitmqApplication.class, args);
    }

}

通过测试,第一条消息在 5s后接收到,第二条消息在 10s后接收到,说明我们的延时队列已经成功

如您在阅读中发现不足,欢迎留言!!!

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/188346.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 苹果Mac允许安装未知来源的应用「建议收藏」

    苹果Mac允许安装未知来源的应用「建议收藏」打开终端,输入命令密码是你的开机密码允许安装未知来源的应用允许sudospctl–master-disable不允许sudospctl–master-enable

    2022年10月22日
  • 2010款汉兰达参数配置_windows安装程序兼容性报告

    2010款汉兰达参数配置_windows安装程序兼容性报告
    关于配置TFS2010的总结:
    TFS2010是我第一次配置,这个东西配置有点麻烦,我还从来没有用过,听经理说,他好像类似于一个大管家,将项目的的需求管理,跟踪,版本管理,测试管理等一系列功能综合在一起,免去了做一个项目要选择许多管理工具的繁琐问题,经理说要不惜一切代价搞定这个东西,在安装的过程中走了很多弯路。
     
    我是参考一篇关于TFS配置的文章配置的,文章介绍的很详细,有插图(我会将它分享给大家),但是我在实际安装的时候还是除了很多问题
    开始我很发愁

  • hive 配置文件以及join中null值的处理「建议收藏」

    hive 配置文件以及join中null值的处理

  • pycharm代码自动提示_pycharm自动整理代码

    pycharm代码自动提示_pycharm自动整理代码那什么,,,,,,是这样的,请先确保你的代码补全功能是打开的。打开操作方式是:file—->powersavemode,把这个前面的√号去掉即可。然后,代码在提示的时候,多打几个字,发现你想要的已经在最上面的时候按tab键即可补全

  • tof相机简介及三维坐标转化,plotly画3D点云[通俗易懂]

    tof相机简介及三维坐标转化,plotly画3D点云[通俗易懂]最近在做TOF相机相关的软件,近年来tof相机开始在手机,车载设备,VR等应用开始增多,产业也开始量化,是一个不错的3维相机的方向。简单介绍一下tof相机吧:TOF是Timeofflight的简写,直译为飞行时间的意思。所谓飞行时间法3D成像,是通过给目标连续发送光脉冲,然后用传感器接收从物体返回的光,通过探测光脉冲的飞行(往返)时间来得到目标物距离。具体原理介绍参考:http://w…

  • 计算机软件师社会需求,java软件工程师的社会环境如何?「建议收藏」

    计算机软件师社会需求,java软件工程师的社会环境如何?「建议收藏」计算机专业成功的改变了很多人的命运,很多人选择计算机培训后拥有了很体面的工作,拥有了很高的工资待遇,整个人的生活水平方面了质的改变。所以越来越多的人希望可以学习计算机,而在专业选择的过程中,JAVA专业吸引到很多求学者的关注。在对JAVA这个专业了解的过程中,JAVA软件工程师的社会环境成为了很多求学者希望了解的内容,这样才能够对这个专业具有更加全面的了解。社会的地位引起关注从JAVA专业成功就业…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号