Rabbitmq的延时队列的使用

配置:

spring:
  rabbitmq:
    addresses: 192.168.108.128:5672
    connection-timeout: 15000
    username: guest
    password: guest
    publisher-confirms: true
    publisher-returns: true

依赖:

  <!--rabbitmq -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
<dependency>    <groupId>org.projectlombok</groupId>    <artifactId>lombok</artifactId></dependency>

配置类:

package com.jds.rabbitmq;

import com.jds.common.constant.QueueEnum;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

@Configuration
public class MQConfig {

    public static final String RBA_QUEUE = "rba.queue";
/*    public static final String QUEUE = "queue";
    public static final String TOPIC_QUEUE1 = "topic.queue1";
    public static final String TOPIC_QUEUE2 = "topic.queue2";
    public static final String HEADER_QUEUE = "header.queue";
    public static final String TOPIC_EXCHANGE = "topicExchage";
    public static final String FANOUT_EXCHANGE = "fanoutxchage";
    public static final String HEADERS_EXCHANGE = "headersExchage";

    *//**
     * Direct模式 交换机Exchange
     * *//*
    @Bean
    public Queue queue() {
        return new Queue(QUEUE, true);
    }

    *//**
     * Topic模式 交换机Exchange
     * *//*
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE1, true);
    }
    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE2, true);
    }
    @Bean
    public TopicExchange topicExchage(){
        return new TopicExchange(TOPIC_EXCHANGE);
    }
    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchage()).with("topic.key1");
    }
    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchage()).with("topic.#");
    }
    *//**
     * Fanout模式 交换机Exchange
     * *//*
    @Bean
    public FanoutExchange fanoutExchage(){
        return new FanoutExchange(FANOUT_EXCHANGE);
    }
    @Bean
    public Binding FanoutBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(fanoutExchage());
    }
    @Bean
    public Binding FanoutBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(fanoutExchage());
    }
    *//**
     * Header模式 交换机Exchange
     * *//*
    @Bean
    public HeadersExchange headersExchage(){
        return new HeadersExchange(HEADERS_EXCHANGE);
    }
    @Bean
    public Queue headerQueue1() {
        return new Queue(HEADER_QUEUE, true);
    }
    @Bean
    public Binding headerBinding() {
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("header1", "value1");
        map.put("header2", "value2");
        return BindingBuilder.bind(headerQueue1()).to(headersExchage()).whereAll(map).match();
    }*/

    /**
     * 订单消息实际消费队列所绑定的交换机
     */
    @Bean
    DirectExchange orderDirect() {
        return (DirectExchange) ExchangeBuilder
                .directExchange(QueueEnum.QUEUE_ORDER_CANCEL.getExchange())
                .durable(true)
                .build();
    }

    /**
     * 订单延迟队列队列所绑定的交换机
     */
    @Bean
    DirectExchange orderTtlDirect() {
        return (DirectExchange) ExchangeBuilder
                .directExchange(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange())
                .durable(true)
                .build();
    }

    /**
     * 订单实际消费队列
     */
    @Bean
    public Queue orderQueue() {
        return new Queue(QueueEnum.QUEUE_ORDER_CANCEL.getName());
    }

    /**
     * 订单延迟队列(死信队列)
     */
    @Bean
    public Queue orderTtlQueue() {
        return QueueBuilder
                .durable(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getName())
                .withArgument("x-dead-letter-exchange", QueueEnum.QUEUE_ORDER_CANCEL.getExchange())//到期后转发的交换机
                .withArgument("x-dead-letter-routing-key", QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey())//到期后转发的路由键
                .build();
    }

    /**
     * 将订单队列绑定到交换机
     */
    @Bean
    Binding orderBinding(DirectExchange orderDirect,Queue orderQueue){
        return BindingBuilder
                .bind(orderQueue)
                .to(orderDirect)
                .with(QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey());
    }

    /**
     * 将订单延迟队列绑定到交换机
     */
    @Bean
    Binding orderTtlBinding(DirectExchange orderTtlDirect,Queue orderTtlQueue){
        return BindingBuilder
                .bind(orderTtlQueue)
                .to(orderTtlDirect)
                .with(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey());
    }

}

发送者:

package com.jds.rabbitmq;

/**
 * @program: red-bag-activity->CancelOrderSender
 * @description:
 * @author: cxy
 * @create: 2019-12-20 17:57
 **/

import com.jds.common.constant.QueueEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 取消订单消息的发出者
 *
 */
@Component
public class CancelOrderSender {
    private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderSender.class);
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendMessage(Long orderId,final long delayTimes){
        //给延迟队列发送消息
        amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderId, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //给消息设置延迟毫秒值
                message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                return message;
            }
        });
        LOGGER.info("send delay message orderId:{}",orderId);
    }
}

接受者:

package com.jds.rabbitmq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 取消订单消息的处理
 */
@Component
@RabbitListener(queues = "mall.order.cancel")
public class CancelOrderReceiver {
    private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderReceiver.class);

    @RabbitHandler
    public void handle(Long orderId){
        System.err.println(System.currentTimeMillis());
        LOGGER.info("receive delay message orderId:{}",orderId);
        System.err.println(System.currentTimeMillis());
        System.err.println(orderId);
        System.out.println("大傻逼222222");
    }
}

配置调用类:

package com.jds.controller;

import com.jds.rabbitmq.CancelOrderSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

/**
 * @program: red-bag-activity->DelayQController
 * @description:
 * @author: cxy
 * @create: 2019-12-20 18:00
 **/
@RestController
public class DelayQController {
    @Autowired
    CancelOrderSender cancelOrderSender;
    @RequestMapping(value = "/id", method = RequestMethod.GET)
    @ResponseBody
    public void detail() {
        long delayTimes = 3 * 1000;
        //发送延迟消息
        cancelOrderSender.sendMessage(12333333L, delayTimes);
        System.out.println(12333333L);
        System.err.println(delayTimes);
        System.out.println(System.currentTimeMillis());

        System.out.println("  大傻逼");

        }
}

测试结果:

2019-12-20 21:48:44.932 |-INFO  [http-nio-8082-exec-1] com.jds.rabbitmq.CancelOrderSender [40] -| send delay message orderId:12333333
12333333
1576849724932
  大傻逼
3000
1576849727975
2019-12-20 21:48:47.975 |-INFO  [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-1] com.jds.rabbitmq.CancelOrderReceiver [21] -| receive delay message orderId:12333333
1576849727975
大傻逼222222
12333333

调用:

http://localhost:8082/id,可以看到时间会延迟三秒。

实现主要地方:

new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //给消息设置延迟毫秒值
                message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                return message;
            }

给参数设置延迟时间

原文地址:https://www.cnblogs.com/xiufengchen/p/12075223.html

时间: 2024-10-29 19:50:41

Rabbitmq的延时队列的使用的相关文章

基于rabbitMQ 消息延时队列方案 模拟电商超时未支付订单处理场景

前言 传统处理超时订单 采取定时任务轮训数据库订单,并且批量处理.其弊端也是显而易见的:对服务器.数据库性会有很大的要求,并且当处理大量订单起来会很力不从心,而且实时性也不是特别好 当然传统的手法还可以再优化一下,即存入订单的时候就算出订单的过期时间插入数据库,设置定时任务查询数据库的时候就只需要查询过期了的订单,然后再做其他的业务操作 jdk延迟队列 DelayQueue 采取jdk自带的延迟队列能很好的优化传统的处理方案,但是该方案的弊.端也是非常致命的,所有的消息数据都是存于内存之中,一旦

RabbitMq 实现延时队列-Springboot版本

rabbitmq本身没有实现延时队列,但是可以通过死信队列机制,自己实现延时队列: 原理:当队列中的消息超时成为死信后,会把消息死信重新发送到配置好的交换机中,然后分发到真实的消费队列: 步骤: 1.创建带有时限的队列 dealLineQueue; 2.创建死信Faout交换机dealLineExchange; 3.创建消费队列realQueue,并和dealLineExchange绑定 4.配置dealLineQueue 的过期时间,消息过期后的死信交换机,重发的routing-key: 以下

RabbitMQ实现延时队列(死信队列)

基于队列和基于消息的TTL TTL是time to live 的简称,顾名思义指的是消息的存活时间.rabbitMq可以从两种维度设置消息过期时间,分别是队列和消息本身. 队列消息过期时间-Per-Queue Message TTL: 通过设置队列的x-message-ttl参数来设置指定队列上消息的存活时间,其值是一个非负整数,单位为微秒.不同队列的过期时间互相之间没有影响,即使是对于同一条消息.队列中的消息存在队列中的时间超过过期时间则成为死信. 死信交换机DLX 队列中的消息在以下三种情况

springboot-rabbitmq:实现延时队列

延时队列应用于什么场景 延时队列顾名思义,即放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费.那么,为什么需要延迟消费呢?我们来看以下的场景 网上商城下订单后30分钟后没有完成支付,取消订单(如:淘宝.去哪儿网)    系统创建了预约之后,需要在预约时间到达前一小时提醒被预约的双方参会    系统中的业务失败之后,需要重试 这些场景都非常常见,我们可以思考,比如第二个需求,系统创建了预约之后,需要在预约时间到达前一小时提醒被预约的双方参会.那么一天之中肯定是会有很多个预约的

spring boot Rabbitmq集成,延时消息队列实现

本篇主要记录Spring boot 集成Rabbitmq,分为两部分, 第一部分为创建普通消息队列, 第二部分为延时消息队列实现: spring boot提供对mq消息队列支持amqp相关包,引入即可: [html] view plain copy <!-- rabbit mq --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-

RabbitMQ:伪延时队列

目录 一.什么是延时队列 二.RabbitMQ实现 三. 延时队列的问题 四.解决RabbitMQ的伪延时方案 ps:伪延时队列先卖个关子,我们先了解下延时队列. 一.什么是延时队列 所谓延时队列是指消息push到队列后,监听的消费者不能第一时间获取消息,需要等到指定时间才能消费. 一般在业务里面需要对某些消息做定时发送,不想走定时任务或者是用户下单之后多长时间自动失效类似的场景可以考虑通过延时队列实现. 二.RabbitMQ实现 MQ本身并不支持直接的延时队列实现,但是我们可以通过Rabbit

springboot使用RabbitMQ实现延时任务

延时队列顾名思义,即放置在该队列里面的消息是不需要立即消费的,而是等待一段时间之后取出消费.那么,为什么需要延迟消费呢?我们来看以下的场景 订单业务: 在电商/点餐中,都有下单后 30 分钟内没有付款,就自动取消订单.短信通知: 下单成功后 60s 之后给用户发送短信通知.失败重试: 业务操作失败后,间隔一定的时间进行失败重试. 本文基于springboot,使用rabbitmq_delayed_message_exchange插件实现延时队列(RabbitMQ及其插件环境安装点此),具体实践如

Rabbitmq 实现延时任务

1.需要用到插件 rabbitmq_delayed_message_exchange 来实现,插件下载地址:https://www.rabbitmq.com/community-plugins.html 2.下载后把插件放到 plugins 里面,然后到 sbin里面打开cmd,执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令 3.插件装好后,重新启动mq,然后集成mq. 首先,导包 <dependency> <

Go RabbitMQ 死信消息队列(二)

实现原理: /** (1)创建一个正常的队列 Q1,目的是处理业务逻辑,比如发送订单消息等 ,对应交换器和绑定键  分别为  E1 和  Bingkey1 (2)创建一个延时消息队列 Q2,设定队列的延时时间为10s,对应的交换器和绑定键分别为 E2和Bingkey2;并在该队列创建时候,设定队列的  (a)超时时间 (b) 超时后跳转的 路由E1和绑定Bingkey1,即超时后跳到     队列Q1上 (3) 将消息先发送到 队列Q2 上,然后等着队列超时,执行逻辑 * 主要测试一个死信队列,