rabbitmq延迟队列demo

工程结构:

定义jar包依赖的版本,版本很重要,rabbit依赖spring,必须一致,否则报错:

<properties>
    <springframework.version>4.2.7.RELEASE</springframework.version>
    <spring-rabbit.version>1.6.1.RELEASE</spring-rabbit.version>
    <junit.version>4.12</junit.version>
</properties>

dependencies:

<dependencies>

    <!-- LOGGING begin -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.5</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-core</artifactId>
        <version>1.0.13</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.0.13</version>
    </dependency>
    <!-- 代码直接调用common-logging会被桥接到slf4j -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>jcl-over-slf4j</artifactId>
        <version>1.7.5</version>
    </dependency>
    <!-- LOGGING end -->

    <!--springframework-->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-core</artifactId>
        <version>${springframework.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</artifactId>
        <version>${springframework.version}</version>
    </dependency>

    <!-- rabbitmq spring依赖 -->
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>${spring-rabbit.version}</version>
    </dependency>

    <!--common utils-->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
        <version>3.3.2</version>
    </dependency>

    <!--test begin-->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>${junit.version}</version>
        <!--<scope>test</scope>-->
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-test</artifactId>
        <version>${springframework.version}</version>
        <!--<scope>test</scope>-->
    </dependency>
    <!--test end-->
</dependencies>

spring-applicationContext:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"

       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
       http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd
         http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="fileEncoding" value="UTF-8"></property>
        <property name="locations">
            <list>
                <value>classpath:applicationContext.properties</value>
            </list>
        </property>
    </bean>

    <context:annotation-config/>

    <bean class="org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor"/>
    <!-- 配置扫描路径 -->
    <context:component-scan base-package="demo"></context:component-scan>

    <!--rabbit server参数 -->
    <rabbit:connection-factory id="connectionFactory"
                               username="${paycenter.mq.user.username}"
                               password="${paycenter.mq.user.password}"
                               addresses="${paycenter.mq.user.host}"></rabbit:connection-factory>

    <import resource="classpath:mq-applicationContext-producer.xml"/>
    <import resource="classpath:mq-applicationContext-consumer.xml"/>
</beans>

mq-applicationContext-producer.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
     http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
     http://www.springframework.org/schema/rabbit
     http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd">

    <!--通过指定下面的admin信息,当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
    <rabbit:admin connection-factory="connectionFactory"/>

    <!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列 -->
    <bean id="mqMessageConverter"
          class="org.springframework.amqp.support.converter.SimpleMessageConverter">
    </bean>

    <!--<bean id="publisherConfirmsReturns" class="com.emaxcard.mq.rabbit.PublisherConfirmsReturns"></bean>-->

    <!--========================延迟队列配置 begin =========================-->
    <rabbit:queue id="agentpayqueryQueue2" durable="true" auto-delete="true" exclusive="false"
                  name="agentpayqueryQueue2"/>
    <rabbit:direct-exchange id="agentpayqueryExchange2" durable="true" auto-delete="true" name="agentpayqueryExchange2">
        <rabbit:bindings>
            <rabbit:binding queue="agentpayqueryQueue2" key="delay"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <rabbit:queue id="agentpayqueryQueue1" durable="true" auto-delete="true" exclusive="false"
                  name="agentpayqueryQueue1">
        <rabbit:queue-arguments>
            <entry key="x-dead-letter-exchange" value="agentpayqueryExchange2"/>
            <entry key="x-message-ttl" value="10000" value-type="java.lang.Long"/>
        </rabbit:queue-arguments>
    </rabbit:queue>
    <rabbit:direct-exchange id="agentpayqueryExchange1" durable="true" auto-delete="true" name="agentpayqueryExchange1">
        <rabbit:bindings>
            <rabbit:binding queue="agentpayqueryQueue1" key="delay"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!--定义RabbitTemplate实例-->
    <!--confirm-callback="publisherConfirmsReturns" return-callback="publisherConfirmsReturns"-->
    <rabbit:template id="agentpayQueryMsgTemplate"
                     exchange="agentpayqueryExchange1" routing-key="delay"
                     connection-factory="connectionFactory" message-converter="mqMessageConverter"
                     mandatory="true"
    />
    <!--========================延迟队列配置 end =========================-->

</beans>

mq-applicationContext-consumer.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
     http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
     http://www.springframework.org/schema/rabbit
     http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd">

    <bean id="agentpayQueryConsumer" class="demo.TestMQConsumer" />

    <!-- TODO 后续删除
    receive-timeout:等待接收超时时长 影响连接创建和销毁

    concurrency:消费者个数
    max-concurrency:最大消费者个数
    min-start-interval:陆续启动  减少并发环境(或是三方系统突然的网络延迟) 大量连接导致的性能耗损
    min-stop-interval:陆续销毁   减少突然的安静 导致大量可用连接被销毁
    min-consecutive-active: 连续N次没有接收发生超时  则认定为需要创建 消费者
    min-consecutive-idle: 连续N次发生了接收超时   则认定消费者需要销毁

    prefetch:每个消费者预读条数 因为异步调用三方 性能瓶颈在网络与三方系统所以预读取条数设置为1(默认为5) 只有一条消息被ACK才会接收下一条消息
    transaction-size:会影响prefetch的数量
    -->
    <!--  监听器 -->
    <!-- queue litener  观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"
                               max-concurrency="20"
                               concurrency="5"
                               prefetch="10">
        <rabbit:listener ref="agentpayQueryConsumer" queue-names="agentpayqueryQueue2" />
    </rabbit:listener-container>
</beans>
Producer类:
package demo;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext.xml")
public class TestMQProducer {

    private static Logger logger = LoggerFactory.getLogger(TestMQProducer.class.getSimpleName());

    @Autowired
    private RabbitTemplate agentpayQueryMsgTemplate;

    @Test
    public void test() throws Exception {
        for (int i = 0; i <= 100; i++) {
            Object data = String.valueOf(i);
            agentpayQueryMsgTemplate.convertAndSend(data);
            logger.info("入队:{}", data);
        }
        Thread.sleep(12000);
    }
}
Consumer类:
package demo;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class TestMQConsumer implements MessageListener {

    private static Logger logger = LoggerFactory.getLogger(TestMQConsumer.class.getSimpleName());

    public void onMessage(Message message) {
        String data = new String(message.getBody());

        try {
            //模拟处理慢
            Thread.sleep(1);

            logger.info("出队:{}", data);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}

至此代码就完毕了。

说明:上面定义队列时我把auto-delete属性设置为true, 所以,当消费者消费完并关闭连接后,队列会自动删除。exchange也如是。(通过mq控制台看,栗子中的agentpayqueryQueue2和agentpayqueryExchange2在执行完就自动消失了,agentpayqueryQueue1和agentpayqueryExchange1还存在。)

spring-rabbit-x.xml里对queue和exchange的auto-delete属性的解释:

Flag indicating that an queue will be deleted when it is no longer in use, i.e. the connection that declared it is closed. Default is false.(rabbit:queue)

Flag indicating that an exchange will be deleted when no longer in use, i.e. the connection that declared it is closed. Default is false.(rabbit:exchange)

消费端的concurrency说明:

同样,看spring-rabbit-x.xml的解释:

The number of concurrent consumers to start for each listener initially.See also ‘max-concurrency‘.

上面我设置的值是5,从mq控制台里看queue的consumer见下图:

从出队日志,可以看出来,共有5个线程在消费这些消息。

原文地址:https://www.cnblogs.com/buguge/p/10110932.html

时间: 2024-10-24 23:18:40

rabbitmq延迟队列demo的相关文章

C# RabbitMQ延迟队列功能实战项目演练

一.需求背景 当用户在商城上进行下单支付,我们假设如果8小时没有进行支付,那么就后台自动对该笔交易的状态修改为订单关闭取消,同时给用户发送一份邮件提醒.那么我们应用程序如何实现这样的需求场景呢?在之前的<C# Redis缓存过期实现延迟通知实战演练>分享课程中阿笨最后总结的时候说过Redis Pub/Sub是一种并不可靠地消息机制,他不会做信息的存储,只是在线转发,那么肯定也没有ack确认机制,另外只有订阅段监听时才会转发!我们是否有更好的方式去实现呢?今天给大家分享的比较好的解决方案就是通过

【RabbitMQ】一文带你搞定RabbitMQ延迟队列

本文口味:鱼香肉丝? ?预计阅读:10分钟 一.说明 在上一篇中,介绍了RabbitMQ中的死信队列是什么,何时使用以及如何使用RabbitMQ的死信队列.相信通过上一篇的学习,对于死信队列已经有了更多的了解,这一篇的内容也跟死信队列息息相关,如果你还不了解死信队列,那么建议你先进行上一篇文章的阅读. 这一篇里,我们将继续介绍RabbitMQ的高级特性,通过本篇的学习,你将收获: 什么是延时队列 延时队列使用场景 RabbitMQ中的TTL 如何利用RabbitMQ来实现延时队列 二.本文大纲

C#实现rabbitmq 延迟队列功能

最近在研究rabbitmq,项目中有这样一个场景:在用户要支付订单的时候,如果超过30分钟未支付,会把订单关掉.当然我们可以做一个定时任务,每个一段时间来扫描未支付的订单,如果该订单超过支付时间就关闭,但是在数据量小的时候并没有什么大的问题,但是数据量一大轮训数据库的方式就会变得特别耗资源.当面对千万级.上亿级数据量时,本身写入的IO就比较高,导致长时间查询或者根本就查不出来,更别说分库分表以后了.除此之外,还有优先级队列,基于优先级队列的JDK延迟队列,时间轮等方式.但如果系统的架构中本身就有

实现rabbitmq 延迟队列功能

最近在研究rabbitmq,项目中有这样一个场景:在用户要支付订单的时候,如果超过30分钟未支付,会把订单关掉.当然我们可以做一个定时任务,每个一段时间来扫描未支付的订单,如果该订单超过支付时间就关闭,但是在数据量小的时候并没有什么大的问题,但是数据量一大轮训数据库的方式就会变得特别耗资源.当面对千万级.上亿级数据量时,本身写入的IO就比较高,导致长时间查询或者根本就查不出来,更别说分库分表以后了.除此之外,还有优先级队列,基于优先级队列的JDK延迟队列,时间轮等方式.但如果系统的架构中本身就有

RabbitMQ 延迟队列实现订单支付结果异步阶梯性通知

在第三方支付中,例如支付宝.或者微信,对于订单请求,第三方支付系统采用的是消息同步返回.异步通知+主动补偿查询的补偿机制. 由于互联网通信的不可靠性,例如双方网络.服务器.应用等因素的影响,不管是同步返回.异步通知.主动查询报文都可能出现超时无响应.报文丢失等情况,所以像支付业务,对结果的通知一般采用几种方案结合的补偿机制,不能完全依赖某一种机制. 例如一个支付结果的通知,一方面会在支付页面跳转时候返回支付结果(一般只用作前端展示使用,非最终状态),同时会采用后台异步通知机制(有前台.后台通知的

Spring Boot(十四)RabbitMQ延迟队列

一.前言 延迟队列的使用场景:1.未按时支付的订单,30分钟过期之后取消订单:2.给活跃度比较低的用户间隔N天之后推送消息,提高活跃度:3.过1分钟给新注册会员的用户,发送注册邮件等. 实现延迟队列的方式有两种: 通过消息过期后进入死信交换器,再由交换器转发到延迟消费队列,实现延迟功能: 使用rabbitmq-delayed-message-exchange插件实现延迟功能: 注意: 延迟插件rabbitmq-delayed-message-exchange是在RabbitMQ 3.5.7及以上

SpringBoot RabbitMQ 延迟队列代码实现

场景 用户下单后,如果30min未支付,则删除该订单,这时候就要可以用延迟队列 准备 利用rabbitmq_delayed_message_exchange插件: 首先下载该插件:https://www.rabbitmq.com/community-plugins.html 然后把该插件放到rabbitmq安装目录plugins下: 进入到sbin目录下,执行"rabbitmq-plugins.bat enable rabbitmq_delayed_message_exchange";

SpringBoot:RabbitMQ 延迟队列

SpringBoot 是为了简化 Spring 应用的创建.运行.调试.部署等一系列问题而诞生的产物,自动装配的特性让我们可以更好的关注业务本身而不是外部的XML配置,我们只需遵循规范,引入相关的依赖就可以轻易的搭建出一个 WEB 工程 初探RabbitMQ消息队列中介绍了RabbitMQ的简单用法,顺带提及了下延迟队列的作用.所谓延时消息就是指当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费. 延迟队列 延迟队列能做什么? 订单业务: 在电商/点餐

RabbitMQ延迟队列简单示例

简介 延迟队列存储的消息是不希望被消费者立刻拿到的,而是等待特定时间后,消费者才能拿到这个消息进行消费.使用场景比较多,例如订单限时30分钟内支付,否则取消,再如分布式环境中每隔一段时间重复执行某操作. 下面举一个简单的例子,例子大概意思是分别在首次发送消息后的10秒.40秒.100秒后重新读取到消息.为了直观,不使用RabbitMQ其他多余的特性. 准备工作 在Centos7下安装RabbitMQ,版本为3.6.12单机版本(非集群),IP是127.0.0.1,端口是15672,使用web管理