RabbitMQ使用延迟插件,代码量直接减少一半!

这篇具有很好参考价值的文章主要介绍了RabbitMQ使用延迟插件,代码量直接减少一半!。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

今天介绍一下使用RabbitMQ的延迟插件方便实现延迟消息的方案。

RabbitMQ 是一个由 Erlang 语言开发的 AMQP(高级消息队列协议) 的开源实现。

RabbitMQ 是轻量级且易于部署的,能支持多种消息协议。

RabbitMQ 可以部署在分布式和联合配置中,以满足高规模、高可用性的需求。

具体特点包括:

  • 可靠性(Reliability):RabbitMQ 使用一些机制来保证可靠性,如持久化、传输确认、发布 确认。

  • 灵活的路由(Flexible Routing):在消息进入队列之前,通过 Exchange 来路由消息的。对 于典型的路由功能,RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能,可以将多个 Exchange 绑定在一起,也通过插件机制实现自己的 Exchange 。

  • 消息集群(Clustering):多个 RabbitMQ 服务器可以组成一个集群,形成一个逻辑 Broker。

  • 高可用(Highly Available Queues):队列可以在集群中的机器上进行镜像,使得在部分节 点出问题的情况下队列仍然可用。

  • 多种协议(Multi-protocol):RabbitMQ 支持多种消息队列协议,比如 STOMP、MQTT等等。

  • 多语言客户端(Many Clients):RabbitMQ 几乎支持所有常用语言,比如 Java、.NET、 Ruby 等等。

  • 管理界面(Management UI):RabbitMQ 提供了一个易用的用户界面,使得用户可以监控 和管理消息 Broker 的许多方面。

  • 跟踪机制(Tracing):如果消息异常,RabbitMQ 提供了消息跟踪机制,使用者可以找出发生 了什么。

  • 插件机制(Plugin System):RabbitMQ 提供了许多插件,来从多方面进行扩展,也可以编 写自己的插件。

RabbitMQ的消息模型

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

Why use rabbitMQ ?

下面,我以一个(花店)商家的角色来向大家形象地举例:

异步

之前顾客来店里下单,我会叫顾客等一下,同时叫店员准备订单,准备好送到顾客手上了顾客才能离开

现在顾客打电话给我:"我要买xxx,地址是:xxx,你帮我送一下"

我拿个小本子记下:顾客a,电话:xxx,地址:xxx

店员有空后就会准备订单并配送

解耦

以前有新订单时,我会亲自找每一个店员(负责准备花束的,负责记账的,负责送花的等),告诉他们有新订单了,有空了处理一下

如果有店员入职,我通知的时候会多通知一个人;离职时,少通知一个人(维护一个需要通知的人员列表)

现在,有新订单的时候,我只需要记到小本子上,店员有空了自己来看

削峰

去年七夕,很多电话打给我,我把每一个订单告诉店员,但是店员忙不过来,客户又一直打电话来催,最后店员累成狗直接罢工

今年七夕节我学乖了,电话打进来我会告诉顾客:"我知道了,会尽快安排处理",然后记到小本子上就行,店员有空时按顺序来处理订单就好

另外还有一种思路,引导客户不一定要在七夕才开始下单,可以提前先买(淘宝的双十一预售就是出于这样的削峰思路)

以上是rabbitMQ解决的核心问题。

How to use rabbitMQ ?

基操安装方式

MAC端

brew install rabbitmq

Windows端

  1. 安装Erlang,下载地址:erlang.org/download/ot…

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

  1. 安装RabbitMQ,下载地址:dl.bintray.com/rabbitmq/al…

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

  1. 安装完成后,进入RabbitMQ安装目录下的sbin目录

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

  1. 在地址栏输入cmd并回车启动命令行,然后输入以下命令启动管理功能:

rabbitmq-plugins enable rabbitmq_management

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

  1. 访问地址查看是否安装成功:http://localhost:15672/

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

CentOS端

安装erlang

# rabbitmq依赖erlang 需要自己去自行下载
cd /path/to/erlang-sound-code && ./configure --prefix=/usr/local/erlang
make && make install 

vim /etc/profile
# 添加
export PATH=$PATH:/usr/local/erlang/bin

source /etc/profile
# 输入erl,会出现版本信息,即安装成功


安装rabbitmq

 #下载 abbitmq_server-3.8.16 并移动到/usr/local/下
vim /etc/profile
 # 添加
export PATH=$PATH:/usr/local/rabbitmq_server-3.8.16/sbin
source /etc/profile

cd /usr/local/rabbitmq_server-3.8.16/sbin 
# 启动
./rabbitmq-server start

功能实现

RabbitMQ实现延迟消息的方式有两种,一种是使用死信队列实现,另一种是使用延迟插件实现。

死信队列的实现网上较多,本文介绍更简单的,使用延迟插件实现(mac环境,java版本)。

另外的安装方式(建议使用这种)

首先准备需要用到的安装文件及插件(rabbitmq_delayed_message_exchange),版本需要匹配,不匹配的版本可能装不上或导致兼容问题。

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式

本人使用的erl_25.0和rabbitMQ-3.10.0(可以到官网下载或者私信作者获取)。使用这种方式安装的优点在于本地安装和服务器安装流程完全一致,不过服务器需要开放安全端口5672,15672视情况,一般建议测试环境开放,生产环境关闭。关注公众号:码猿技术专栏,回复关键词:1111 获取阿里内部Java性能调优手册

安装erl和rabbitMQ,具体步骤略(这个应该没人不会吧,逃~)。

将插件文件复制到RabbitMQ安装目录的plugins目录下,执行以下命令后重启rabbitMQ:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

实现延迟消息

以一个实际业务场景举例:当客服状态为在线且3分钟未回复客户消息时,自动重启im会话机器人接管会话。这是一个常见的延迟消息使用场景。

首先在pom.xml文件中添加AMQP相关依赖

<!--消息队列相关依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml添加RabbitMQ的相关配置

spring:
  rabbitmq:
    host: localhost # rabbitmq的连接地址
    port: 5672 # rabbitmq的连接端口号
    virtual-host: /mall # rabbitmq的虚拟host
    username: im # rabbitmq的用户名
    password: xxxxxx # rabbitmq的密码
    publisher-confirms: true #如果对异步消息需要回调必须设置为true

接下来创建RabbitMQ的java配置,主要用于配置交换机、队列和绑定关系

/**
 * 消息队列配置
 */
@Configuration
public class RabbitMqConfig {
    /**
     * 机器人消息重启插件消息队列所绑定的交换机
     */
    @Bean
    CustomExchange chatPluginDirect() {
        //创建一个自定义交换机,可以发送延迟消息
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getExchange(), "x-delayed-message", true, false, args);
    }

    /**
     * 机器人消息重启插件队列
     */
    @Bean
    public Queue chatPluginQueue() {
        return new Queue(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getName());
    }

    /**
     * 将机器人消息重启插件队列绑定到交换机
     */
    @Bean
    public Binding chatPluginBinding(CustomExchange chatPluginDirect, Queue chatPluginQueue) {
        return BindingBuilder
                .bind(chatPluginQueue)
                .to(chatPluginDirect)
                .with(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getRouteKey())
                .noargs();
    }
}

创建一个消息的发出者,通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间

/**
 * 机器人重启队列发出者
 */
@Component
@Slf4j
public class ChatQueueSender {
    private static Logger LOGGER = LoggerFactory.getLogger(ChatQueueSender.class);
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendMessageToChat(Long cmid, final long delayTimes) {
        //给延迟队列发送消息
        amqpTemplate.convertAndSend(QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getExchange(), QueueEnum.QUEUE_RESET_MESSAGE_CANCEL.getRouteKey(), cmid, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //给消息设置延迟毫秒值
                message.getMessageProperties().setHeader("x-delay", delayTimes);
                return message;
            }
        });
    }
}

创建一个消息的接收者,用于处理延迟插件队列中的消息。

/**
 * 机器人重启队列处理者
 */
@Component
@Slf4j
@RabbitListener(queues = "im.chat.cancel")
public class ChatQueueReceiver {
    @Autowired
    private ChatRestartRobotService chatRestartRobotService;

    @RabbitHandler
    public void handleOnChat(Long cmid) {
//        log.info("机器人会话重启");
        chatRestartRobotService.restartRobot(cmid);
    }
}

最后,在对应的地方调用即可:

rabbitmq延迟队列插件,java-rabbitmq,rabbitmq,分布式文章来源地址https://www.toymoban.com/news/detail-706037.html

 
                    

到了这里,关于RabbitMQ使用延迟插件,代码量直接减少一半!的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • RabbitMQ+springboot用延迟插件实现延迟消息的发送

    延迟队列:其实就是死信队列中消息过期的特殊情况 延迟队列应用场景: 可以用死信队列来实现,不过死信队列要等上一个消息消费成功,才会进行下一个消息的消费,这时候就需要用到延迟插件了,不过要线在docker上装一个插件 前置条件是在Docker中部署过RabbitMq。 1、打开

    2024年02月10日
    浏览(50)
  • [超详细]RabbitMQ安装延迟消息插件

    Community Plugins — RabbitMQ https://www.rabbitmq.com/community-plugins.html 进入以上地址以后,找到Routing里边的rabbitmq_delayed_message_exchange然后点击Releases   下载完成以后  然后解压到plugins文件中  然后再sbin目录下运行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange  查看交换机类型中是否有

    2024年02月07日
    浏览(47)
  • liunx+docker+rabbitmq安装延迟队列插件

    前言 在这篇文章中,我们将讨论如何在 Linux 系统上安装 Docker 和 RabbitMQ,并设置延迟队列。 Docker 是一个开放源代码的软件,它可以使应用程序的部署更加简单,而 RabbitMQ 是一个开放源代码的消息代理软件,它接受和转发消息。 延迟队列是一种在特定的延迟之后才开始处理

    2024年02月11日
    浏览(47)
  • Docker版RabbitMQ安装延迟队列插件及延迟队列项目应用实战

    在项目中经常有延迟业务处理的背景,此时可以借助于Rabbitmq的延迟队列进行实现,但Rabbitmq本身并不支持延迟队列,但可以通过安装插件的方式实现延迟队列 首先确认目前项目使用的Rabbitmq的版本,这里博主的版本是3.9.15的。 访问 Rabbitmq的github网址,检索 delay 找到插件 rabb

    2024年02月02日
    浏览(43)
  • RabbitMQ延迟插件【安装windows/Linux(Docker)】

    死信队列: 就是正常消费者在一定时间内没有进行消费的消息会发送到死信队列中,而会有消费者在监听死信队列。消费者在 消息被拒绝、消息TTL过期、队列达到最大长度 这三种情况下 ,消息会被消费者转发到死信队列。 延迟队列: 其实就是死信队列中消息过期的特殊情

    2024年02月09日
    浏览(60)
  • Docker中为RabbitMQ安装rabbitmq_delayed_message_exchange延迟队列插件

    1、前言 rabbitmq_delayed_message_exchange是一款向RabbitMQ添加延迟消息传递(或计划消息传递)的插件。 插件下载地址:https://www.rabbitmq.com/community-plugins.html 1、下载插件 首先需要确定我们当前使用的RabbitMQ的版本,我们可以直接登录Web端的管理界面查看版本   也可以在RabbitMQ容器中

    2024年02月12日
    浏览(49)
  • Linux安装rabbitMq RPM安装 以及带延迟插件

    文档中rabbitmq下载链接 以及延迟插件 网盘下载 目前下载文件中版本已经过多个服务器安装测试 完全成功 rpm -ivh openssl-libs-1.0.2k-19.el7.x86_64.rpm --force --nodeps rpm -ivh libnsl-2.34-28.el9_0.x86_64.rpm --force --nodeps rpm -ivh erlang-23.3-2.el7.x86_64.rpm --force --nodeps rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm --fo

    2024年02月03日
    浏览(46)
  • RabbitMQ-延迟队列的使用

    目录 一、使用场景  二、第一种方式:创建具有超时功能且绑定死信交换机的消息队列 三、第二种方式:创建通用延时消息 四、第三种方式:使用rabbitmq的延时队列插件,实现同一个队列中有多个不同超时时间的消息,并按时间超时顺序出队 4.1 下载延迟插件 4.2 安装插件

    2024年04月14日
    浏览(37)
  • RabbitMQ系列(27)--RabbitMQ使用Federation Exchange(联邦交换机)解决异地访问延迟问题

    前言: (broker北京)、(broker深圳)彼此之间相距甚远,网络延迟是一个不得不面对的问题。有一个在北京的业务(Client北京)需要连接(broker北京),向其中的交换器exchangeA发送消息,此时的网络延迟很小,(Client北京)可以迅速将消息发送至exchangeA 中,就算在开启了publisherconfirm机制或

    2024年02月13日
    浏览(71)
  • 【RabbitMQ】 RabbitMQ 消息的延迟 —— 深入探索 RabbitMQ 的死信交换机,消息的 TTL 以及延迟队列

    消息队列是现代分布式应用中的关键组件,用于实现异步通信、解耦系统组件以及处理高并发请求。消息队列可以用于各种应用场景,包括任务调度、事件通知、日志处理等。在消息队列的应用中,有时需要实现消息的延迟处理、处理未能成功消费的消息等功能。 本文将介绍

    2024年02月05日
    浏览(79)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包