精华内容
下载资源
问答
  • 2022-04-12 12:01:23

    1、在启用了 rabbitmq management plugin 之后,可以通过 http://IP:15672/cli/ 下载 rabbitmqadmin shell 文件,linux/windows都可用curl,如下:

    ##下载文件
    sudo curl http://IP:15672/cli/rabbitmqadmin -o /usr/sbin/rabbitmqadmin
    
    ##添加可执行权限
    sudo chmod a+x /usr/sbin/rabbitmqadmin
    
    
    ##示例:
    sudo curl http://192.168.16.190:15672/cli/rabbitmqadmin -o /usr/sbin/rabbitmqadmin 
    sudo chmod 755 /usr/sbin/rabbitmqadmin
    

    此时就可以通过 rabbitmqadmin 删除队列,
    在这里插入图片描述
    2、删除队列名含trdL字符串的队列

    rabbitmqadmin list queues name | awk '{print $2}' | grep "trdL" | xargs -I qn rabbitmqadmin delete queue name=qn
    

    备注:rabbitmqadmin使用python作为脚本语言,所以需要确保主机环境安装了python 2.6以上。具体见 http://www.rabbitmq.com/management-cli.html。

    更多相关内容
  • RabbitMQ消息队列

    千次阅读 2021-12-09 23:30:06
    MQ(message queue),从字面意思上看,本质是个队列,FIFO 先入先出,只不过队列中存放的内容是message 而已,还是一种跨进程的通信机制,用于上下游传递消息。在互联网架构中,MQ 是一种非常常见的上下游“逻辑解耦+...

    1 MQ的相关概念

    1.1 什么是MQ

    MQ(message queue),从字面意思上看,本质是个队列,FIFO 先入先出,只不过队列中存放的内容是message 而已,还是一种跨进程的通信机制,用于上下游传递消息。在互联网架构中,MQ 是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了 MQ 之后,消息发送上游只需要依赖 MQ,不用依赖其他服务。

    1.2 为什么要用MQ

    1 流量削峰

    举个例子,如果订单系统最多能处理一万次订单,这个处理能力应付正常时段的下单时绰绰有余,正常时段我们下单一秒后就能返回结果。但是在高峰期,如果有两万次下单操作系统是处理不了的,只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲,我们可以取消这个限制,把一秒内下的订单分散成一段时间来处理,这时有些用户可能在下单十几秒后才能收到下单成功的操作,但是比不能下单的体验要好。

    2 应用解耦

    以电商应用为例,应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后,如果耦合调用库存系统、物流系统、支付系统,任何一个子系统出了故障,都会造成下单操作异常。当转变成基于消息队列的方式后,系统间调用的问题会减少很多,比如物流系统因为发生故障,需要几分钟来修复。在这几分钟的时间里,物流系统要处理的内存被缓存在消息队列中,用户的下单操作可以正常完成。当物流系统恢复后,继续处理订单信息即可,中单用户感受不到物流系统的故障,提升系统的可用性。

    在这里插入图片描述

    3 异步处理

    有些服务间调用是异步的,例如 A 调用 B,B 需要花费很长时间执行,但是 A 需要知道 B 什么时候可以执行完,以前一般有两种方式,A 过一段时间去调用 B 的查询 api 查询。或者 A 提供一个 callback api, B 执行完之后调用 api 通知 A 服务。这两种方式都不是很优雅,使用消息总线,可以很方便解决这个问题,A 调用 B 服务后,只需要监听 B 处理完成的消息,当 B 处理完成后,会发送一条消息给 MQ,MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api,也不用提供 callback api。同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息。
    在这里插入图片描述

    1.3 MQ的分类

    1. ActiveMQ

    • 优点

      • 单机吞吐量万级,时效性 ms 级,可用性高,基于主从架构实现高可用性,消息可靠性较低的概率丢失数据。
    • 缺点

      • 官方社区现在对 ActiveMQ 5.x 维护越来越少,高吞吐量场景较少使用。

    2. Kafka

    大数据的杀手锏,谈到大数据领域内的消息传输,则绕不开Kafka,这款为大数据而生的消息中间件,以其百万级TPS的吞吐量名声大噪,迅速成为大数据领域的宠儿,在数据采集、传输、存储的过程中发挥举足轻重的作用。目前已经被LinkedIn,Uber,Twitter,Netfix等大公司所采纳。

    • 优点

      • 性能卓越,单机写入 TPS 约在百万条/秒,最大的优点,就是吞吐量高。时效性 ms 级可用性非常高,kafka 是分布式的,一个数据多个副本,少数机器宕机,不会丢失数据,不会导致不可用,消费者采用 Pull 方式获取消息, 消息有序, 通过控制能够保证所有消息被消费且仅被消费一次;有优秀的第三方Kafka Web 管理界面 Kafka-Manager;在日志领域比较成熟,被多家公司和多个开源项目使用;功能支持:功能较为简单,主要支持简单的 MQ 功能,在大数据领域的实时计算以及日志采集被大规模使用
    • 缺点

      • Kafka 单机超过 64 个队列/分区,Load 会发生明显的飙高现象,队列越多,load 越高,发送消息响应时间变长,使用短轮询方式,实时性取决于轮询间隔时间,消费失败不支持重试;支持消息顺序,但是一台代理宕机后,就会产生消息乱序,社区更新较慢

    3. RocketMQ

    RocketMQ 出自阿里巴巴的开源产品,用 Java 语言实现,在设计时参考了 Kafka,并做出了自己的一些改进。被阿里巴巴广泛应用在订单,交易,充值,流计算,消息推送,日志流式处理,binglog 分发等场
    景。

    • 优点

      • 单机吞吐量十万级,可用性非常高,分布式架构,消息可以做到 0 丢失,MQ 功能较为完善,还是分布式的,扩展性好,支持 10 亿级别的消息堆积,不会因为堆积导致性能下降,源码是 java 我们可以自己阅读源码,定制自己公司的 MQ
    • 缺点

      • 支持的客户端语言不多,目前是 java 及 c++,其中 c++不成熟;社区活跃度一般,没有在 MQ核心中去实现 JMS 等接口,有些系统要迁移需要修改大量代码

    4. RabbitMQ

    2007 年发布,是一个在 AMQP(高级消息队列协议)基础上完成的,可复用的企业消息系统,是当前最主流的消息中间件之一。

    • 优点

      • 由于 erlang 语言的高并发特性,性能较好;吞吐量到万级,MQ 功能比较完备,健壮、稳定、易用、跨平台、支持多种语言 如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持 AJAX 文档齐全;开源提供的管理界面非常棒,用起来很好用,社区活跃度高;更新频率相当高
    • 缺点

      • 商业版需要收费,学习成本较高

    1.4 MQ的选择

    1. kafka

    Kafka 主要特点是基于 Pull 的模式来处理消息消费,追求高吞吐量,一开始的目的就是用于日志收集和传输,适合产生大量数据的互联网服务的数据收集业务。大型公司建议可以选用,如果有日志采集功能,肯定是首选 kafka 了。

    2. RocketMQ

    天生为金融互联网领域而生,对于可靠性要求很高的场景,尤其是电商里面的订单扣款,以及业务削峰,在大量交易涌入时,后端可能无法及时处理的情况。RoketMQ 在稳定性上可能更值得信赖,这些业务场景在阿里双 11 已经经历了多次考验,如果你的业务有上述并发场景,建议可以选择 RocketMQ。

    3. RabbitMQ

    结合 erlang 语言本身的并发优势,性能好时效性微秒级,社区活跃度也比较高,管理界面用起来十分方便,如果你的数据量没有那么大,中小型公司优先选择功能比较完备的 RabbitMQ。

    1.5 RabbitMQ

    RabbitMQ概念

    RabbitMQ 是一个消息中间件:它接受并转发消息。你可以把它当做一个快递站点,当你要发送一个包裹时,你把你的包裹放到快递站,快递员最终会把你的快递送到收件人那里,按照这种逻辑 RabbitMQ 是一个快递站,一个快递员帮你传递快件。RabbitMQ 与快递站的主要区别在于,它不处理快件而是接收,存储和转发消息数据。

    四大核心概念

    • 生产者

      • 产生数据发送消息的程序是生产者
    • 交换机

      • 交换机是RabbitMQ非常重要的一个部件,一方面它接收来自生产者的消息,另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息,是将这些消息推送到特定队列还是推送到多个队列,亦或者是把消息丢弃,这个得由交换机类型决定
    • 队列

      • 队列是RabbitMQ内部使用的一种数据结构,尽管消息流经RabbitMQ和应用程序,但他们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束,本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列,许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式
    • 消费者

      • 消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者,消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又可以是消费者。

    名词解释

    在这里插入图片描述

    • Broker

      • 接收和分发消息的应用,RabbitMQ Server就是Message Broker
    • Virtual host

      • 处于多租户和安全因素设计的,把AMQP的基本组件划分到一个虚拟的分组中,类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ Server提供的服务时,可以划分出多个vhost,每个用户在自己的vhos创建exchange/queue等。
    • Connection

      • producer/consumer和broker之间的TCP连接
    • Channel

      • 如果每一次访问RabbitMQ都建立一个Connection,在消息量大的时候建立TCP Connection的开销是巨大的,效率也较低。Channel是在Connection内部建立的逻辑连接,如果应用程序支持多线程,通常每个thread创建单独的channel进行通讯,AMQP method包含了channel id帮助客户端和message broker识别channel,所以channel之间是完全隔离的。channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。
    • Exchange

      • message到达broker的第一站,根据分发规则,匹配查询表中的routing key,分发消息到queue中去。常用的类型有direct(point-to-point)topic(publish-subscribe)andfanout(multicast)
    • Queue

      • 消息最终被送到这里等待consumer取走
    • Binding

      • exchange和queue之间的虚拟连接,binding中可以包含routing key,Binding信息被保存到exchange中的查询表中,用于message的分发依据。

    1.6 RabbitMQ安装(linux)

    1. 下载

    2. 文件上传
      将文件上传到/usr/local/software目录下在这里插入图片描述

    3. 安装文件(分别按照以下顺序安装)

      • rpm -ivh erlang-21.3-1.el7.x86_64.rpm
      • yum install socat -y
      • rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm
    4. 常用命令

    • 添加开机启动RabbitMQ服务

      • chkconfig rabbitmq-server on
    • 启动服务

      • /sbin/service rabbitmq-server start
    • 查看服务状态

      • /sbin/service rabbitmq-server status
      • 在这里插入图片描述
    • 停止服务

      • /sbin/service rabbitmq-server stop
    • 开启 web 管理插件

      • rabbitmq-plugins enable rabbitmq_management
      • 需先关闭防火墙

    若开启web插件是报错
    在这里插入图片描述
    解决办法:
    1. 配置根路径下/etc/hosts 文件
    2. vim hosts
    3. i 进入编辑模式
    4. 在里面添加
    5. 127.0.0.1 xxxx (xxxx 为当前hostname @后面的就是,例如我的就是PumpkinOS)在这里插入图片描述
    6. :wq 保存退出,再次开启即可

    • 添加一个新的用户
      • 创建账号

        • rabbitmqctl add_user pumpkin 123456
      • 设置用户角色

        • rabbitmqctl set_user_tags pumpkin administrator
      • 设置用户权限

        • 语法 set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
        • rabbitmqctl set_permissions -p "/" pumpkin".*" ".*" ".*"
        • 用户 pumpkin 具有/vhost1 这个 virtual host 中所有资源的配置、写、读权限

    2. Hello World

    使用Java编写两个程序。发送单个消息的生产者和接收消息并打印出来的消费者。我将介绍Java API的一些细节。
    在下图中,P是生产者,C是消费者。中间的框是一个队列,代表使用者保留的消息缓冲区。
    在这里插入图片描述

    引入依赖

    <dependencies>
    	 <!--rabbitmq 依赖客户端-->
    	 <dependency>
    		 <groupId>com.rabbitmq</groupId>
    		 <artifactId>amqp-client</artifactId>
    		 <version>5.8.0</version>
    	 </dependency>
    	 <!--操作文件流的一个依赖-->
    	 <dependency>
    		 <groupId>commons-io</groupId>
    		 <artifactId>commons-io</artifactId>
    		 <version>2.6</version>
    	 </dependency>
    </dependencies>
    

    消息生产者

    /**
     * 生产者
     */
    public class Producer {
        public static final String QUEUE_NAME = "hello" ;
    
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建一个连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //设置工厂ip以连接RabbitMQ的队列
            factory.setHost("192.168.88.128");
            //设置用户名
            factory.setUsername("pumpkin");
            //设置密码
            factory.setPassword("123456");
            //创建连接
            Connection connection = factory.newConnection();
            //产生信道
            Channel channel = connection.createChannel();
            /**
             * 生成一个队列
             * 参数列表
             * 1. 队列名称
             * 2. 队列里面的消息是否持久化,默认为false即非持久化
             * 3. 该队列是否只供一个消费者进行消费(是否进行消息共享),true可以多个消费者消费,false只能一个消费者消费
             * 4. 是否自动删除 最后一个消费者断开连接后,该队列是否自动删除 true自动删除 false不自动删除
             * 5. 其他参数
             */
            channel.queueDeclare(QUEUE_NAME , false , false ,false , null) ;
            //发消息
            String message = "hello world2" ;
            /**
             * 发送消息
             * 参数列表
             * 1. 表示发送到哪个交换机
             * 2. 路由的key值,这里是队列名
             * 3. 其他参数信息
             * 4. 发送的消息
             */
            channel.basicPublish("" , QUEUE_NAME , null , message.getBytes());
            System.out.println("消息发送完毕");
        }
    }
    

    消息消费者

    /**
     * @author pumpkin
     * @date 2021/12/7 0007 下午 12:34
     */
    
    //消费者
    public class Consumer {
        public static final String QUEUE_NAME = "hello" ;
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.88.128");
            factory.setUsername("pumpkin");
            factory.setPassword("123456");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            /**
             * 消费者消费消息
             * 参数列表
             * 1. 消费哪个队列
             * 2. 消费成功之后是否要自动应答 true代表的自动应答 false代表手动应答
             * 3. 消费者未成功消费的回调
             * 4. 消费者取消消费的回调
             */
            DeliverCallback deliverCallback = (consumerTag , message)->{
                System.out.println(new String(message.getBody()));
            };
    
            CancelCallback cancelCallback = consumerTag->{
                System.out.println("消息消费被中断");
            };
    
            channel.basicConsume(QUEUE_NAME ,true ,deliverCallback ,cancelCallback) ;
        }
    }
    

    3 Work Queues

    工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务,而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时,这些工作线程将一起处理这些任务。

    3.1 轮询分发消息

    在这个案例中,我会启动两个消息消费者线程,一个消息生产者线程。

    抽取工具类

    /**
     * @author pumpkin
     * @date 2021/12/7 0007 下午 14:12
     */
    public class RabbitmqUtils {
    
        public static Channel getChannel() throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.88.128");
            factory.setUsername("pumpkin");
            factory.setPassword("123456");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            return channel ;
        }
    
    }
    

    消费者代码

    /**
     * @author pumpkin
     * @date 2021/12/7 0007 下午 14:25
     */
    public class Worker01 {
        public static final String QUEUE_NAME = "hello" ;
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Channel channel = RabbitmqUtils.getChannel();
            DeliverCallback deliverCallback = (consumerTag , message)->{
                System.out.println("消息接收: " + new String(message.getBody()));
            };
            CancelCallback cancelCallback = consumerTag->{
                System.out.println(consumerTag + "消息取消接收");
            };
            System.out.println("c1等待接收消息........");
            channel.basicConsume(QUEUE_NAME , true , deliverCallback ,cancelCallback);
        }
    }
    
    

    开启多线程运行Worker01代码。启动两个消费者线程

    在这里插入图片描述
    在这里插入图片描述

    启动一个生产者线程

    /**
     * @author pumpkin
     * @date 2021/12/7 0007 下午 14:45
     */
    public class Task01 {
        public static final String QUEUE_NAME = "hello" ;
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Channel channel = RabbitmqUtils.getChannel();
            channel.queueDeclare(QUEUE_NAME , false , false ,false , null) ;
            Scanner scanner = new Scanner(System.in);
            while( scanner.hasNext() ){
                String message = scanner.next();
                channel.basicPublish("" , QUEUE_NAME , null , message.getBytes());
                System.out.println("消息发送完毕" + message);
            }
        }
    }
    

    发送消息
    在这里插入图片描述
    C2接收消息
    在这里插入图片描述
    C2接收消息
    在这里插入图片描述
    通过程序执行发现生产者总共发送 4 个消息,消费者 1 和消费者 2 分别分得两个消息,并且是按照有序的一个接收一次消息

    3.2 消息应答

    3.2.1 概念

    消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况。RabbitMQ一旦向消费者传递了一条消息,便立即将消息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将会丢失正在处理的消息。以及后续发送给该消费者的消息,因为它无法接收到。
    为了保证消息在发送过程中不丢失,RabbitMQ引入消息应答机制,消息应答就是:消费者在接收到消息并且处理该消息之后,告诉RabbitMQ它已经处理了,RabbitMQ可以把该消息删除了。

    3.2.2 自动应答

    消息发送后立即被认为已经传送成功,这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前,消费者那边出现连接或者channel关闭,那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息,没有对传递的消息数量进行限制,当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终使得内存耗尽,最终这些消费者线程被操作系统杀死,所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用

    3.2.3 消息应答的方法

    • Channel.basicAck(用于肯定确认)

      • RabbitMQ已经知道该消息并且成功的处理消息,可以将其丢弃了
    • Channel.basicNack(用于否定确认)

    • Channel.basicReject(用于否定确认)

      • 与Channel.basicNack相比少一个参数
      • 不处理该消息了,直接拒绝,可以将其丢弃了

    3.2.4 Multiple的解释

    手动应答的好处是可以批量应答并且减少网络拥堵
    在这里插入图片描述
    multiple的true和false代表不同的意思

    • true 代表批量应答channel上未应答的消息

      • 比如说channel上有传送tag的消息5,6,7,8。当前tag是8那么此时5~8的这些还未应答的消息都会被确认收到消息应答。
    • false

      • 同true相比,只会应答tag=8的消息,5,6,7这三个消息依然不会被确认收到消息应答

    在这里插入图片描述

    3.2.5 消息自动重新入队

    如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或TCP连接丢失),导致消息未发送ACK确认,RabbitMQ将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者可以处理,他将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确保不会丢失任何消息。
    在这里插入图片描述

    3.2.6 消息手动应答

    默认消息采用的是自动应答,所以要想实现消息消费过程种不丢失,需要把自动应答改为手动应答。

     DeliverCallback deliverCallback = (consumerTag , message)->{
                try {
                    Thread.sleep(1000*1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("接收消息:" + new String(message.getBody() , "UTF-8"));
                /**
                 * 手动应答
                 * 参数列表
                 * 1. 消息的标记
                 * 2. 是否批量应答 false:不批量应答信道中的消息  true:批量
                 */
                channel.basicAck(message.getEnvelope().getDeliveryTag() , false);
            };
    
            /**
             * 消费者消费消息
             * 参数列表
             * 1. 消费哪个队列
             * 2. 消费成功之后是否要自动应答 true代表的自动应答 false代表手动应答
             * 3. 消费者未成功消费的回调
             * 4. 消费者取消消费的回调
             */
            channel.basicConsume(ACK_QUEUE,false, deliverCallback, consumerTag->{
                System.out.println("消息放弃接收");
            }) ;
    

    3.2.7 手动应答案例

    正常情况下,消息发送方发送两个消息C1和C2分别接收到消息并处理。
    发送方Task02发送两个消息
    在这里插入图片描述
    两个消费者Worker03和Worker04分别处理两个消息
    在这里插入图片描述
    在这里插入图片描述
    在发送放发送消息bb,发送消息之后把Worker04消费者停掉,按理说该Worker04来处理该消息,但是由于它处理时间较长,在还未处理完,也就是说Worker04还没有执行ack代码的时候,Worker04被停掉了,此时会看到消息被Worker03接收到了,说明消息bb被重新入队,然后分配给能处理消息的Worker03处理了。
    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述

    3 RabbitMQ持久化

    3.3.1 概念

    刚刚我们已经看到了如何处理任务不丢失的情况,但是如何保障当RabbitMQ服务停掉以后消息生产者发送过来的消息不丢失。默认情况下RabbitMQ退出或由于某种原因崩溃时,它忽视队列和消息,除非告知它不要这么做。确保消息不会丢失需要做两件事:

    • 将队列标记为持久化
    • 将消息标记为持久化

    3.3.2 队列如何持久化

    之前创建的队列都是非持久化的,rabbitmq如果重启的话,该队列就会被删除,如果要队列实现持久化,需要在声明队列的时候把durable参数设置为持久化

      /**
             * 生成一个队列
             * 参数列表
             * 1. 队列名称
             * 2. 队列里面的消息是否持久化,默认为false即非持久化
             * 3. 该队列是否只供一个消费者进行消费(是否进行消息共享),true可以多个消费者消费,false只能一个消费者消费
             * 4. 是否自动删除 最后一个消费者断开连接后,该队列是否自动删除 true自动删除 false不自动删除
             * 5. 其他参数
             */
            channel.queueDeclare(QUEUE_NAME , true , false ,false , null) ;
    

    但是需要注意,如果之前声明了同名的队列且不是持久化的,需要把原先的队列删除,或者重现创建一个持久化的队列,不然会出现错误

    在这里插入图片描述
    以下为控制台中持久化与非持久化队列的UI显示区别

    在这里插入图片描述
    这个时候即使重启rabbitmq服务器,comfirm_queue队列依然存在。

    3.3.3 消息实现持久化

    要想让消息实现持久化需要在消息生产者发送消息时,在basicPublish中添加MessageProperties.PERSISTENT_TEXT_PLAIN添加这个属性。

     /**
             * 发送消息
             * 参数列表
             * 1. 表示发送到哪个交换机
             * 2. 路由的key值,这里是队列名
             * 3. 其他参数信息
             * 4. 发送的消息
             */
            channel.basicPublish("" , QUEUE_NAME , MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    

    将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘,但是这里依然存在当消息刚准备存储在磁盘的时候,但是还没有存储完,消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强,但是对于简单任务队列而言,这已经绰绰有余了。

    3.3.4 不公平分发

    在最开始的时候说到了RabbitMQ分发消息采用的轮询分发,但是在某种场景下这种策略并不是很好,比方说有两个消费者在处理任务,其中有个消费者1处理任务的速度非常快,而另外一个消费者2处理速度却很慢,这个时候如果还是采用轮询分发的话就会导致处理速度快的这个消费者很大一部分时间处于空闲状态,而处理慢的那个消费者一直都在干活,这种分配方式在这种情况下就不太好,但是RabbitMQ并不知道这种情况,它依然很公平的进行分发。

    为了避免这种情况,可以设置参数

    int prefetchCount = 1 ;
    channel.basicQos(prefetchCout) ;
    

    在这里插入图片描述

    在这里插入图片描述
    意思就是如果这个任务还没有处理完或者还没有应答,你就先别分配给我,我目前只能处理一个任务,然后rabbitmq就会把该任务分配给没有那么忙的那个空闲消费者,当然如果所有的消费者都没有完成手上的任务,队列还在不停的添加新任务,队列有可能就会遇到队列被撑满的情况,这个时候就只能添加新的worker或者改变其他存储任务的策略。

    3.3.5 预取值

    本身消息的发送就是异步发送的,所以在任何时候,channel上肯定不止只有一个消息,另外来自消费者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区,因此希望开发人员能限制此缓冲区的大小,以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用basic.qos方法设置“预取计数”来完成。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量,RabbitMQ将停止在通道上传递更多消息,除非至少有一个未处理的消息被确认,例如,假设在通道上有未确认的消息5,6,7,8,并且通道的预取计数设置为4,此时RabbitMQ将不会在该通道上再传递任何消息,除非至少有一个未应答的消息被ack。比方说tag=6这个消息刚刚被确认ACK,RabbitMQ将会感知这个情况并再发送一条消息。消息应答和Qos预取值对用户吞吐量有重大影响。通常,增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率时最佳的,但是,在这种情况下已传递但尚未处理的消息的数量也会增加,从而增加了消费者的RAM消耗(随机存取存储器)应该小心使用具有无限预处理的自动确认模式或手动确认模式,消费者消费了大量的消息,如果没有确认的话,会导致消费者连接节点的内存消耗变大,所以找到合适的预取值是一个反复实验的过程,不同的负载该值取值也不同,100到300范围内的值通常可提供最佳的吞吐量,并且不会给消费者带来太大的风险。预取值为1是最保守的。当然这将使吞吐量变得很低,特别是消费者连接延迟很严重的情况下,特别是在消费者连接等待时间较长的环境中,对于大多数应用来说,稍微高一点的值是最佳的。

    在这里插入图片描述

    4 发布确认

    TODO

    展开全文
  • 手动创建队列 rabbitmqadmin declare queue name=test auto_delete=...2.AutoDelete: 是否自动删除, true表示自动删除, false为非自动删除 补充: 如需指定vhost的队列(默认是/) 加上 --vhost rabbitmqadmin li..
    • 手动创建队列

    rabbitmqadmin declare queue name=test  auto_delete=false durable=false  --username=XXX  --password=XXX
    

    123
    1.Durable: 是否需要持久化, true表示持久化, false为非持久化
    2.AutoDelete: 是否自动删除, true表示自动删除, false为非自动删除
    补充:
    如需指定vhost的队列(默认是/) 加上 --vhost

    rabbitmqadmin list queues指令,可以查看所创建的队列信息。

    查看命令帮助信息:
    python rabbitmqadmin help subcommands

    rabbitmqctl add_user 用户 密码
    查看用户:rabbitmqctl list_users
    给用户赋权:
    rabbitmqctl set_permissions -p / 用户 ‘.’ '.’ ‘.*’
    查看用户权限:rabbitmqctl list_user_permissions 用户
    设置用户登录web界面权限:
    rabbitmqctl set_user_tags 用户 (administrator,monitoring, management)

    展开全文
  • 死信队列 死信的概念 先从概念解释上搞清楚这个定义,死信,顾名思义就是无法被消费的消息...应用场景:为了保证订单业务的消息数据不丢失,需要使用到 RabbitMQ 的死信队列机制,当消息消费发生异常时,将消息投入死信

    死信队列

    死信的概念

    先从概念解释上搞清楚这个定义,死信,顾名思义就是无法被消费的消息,字面意思可以这样理解,一般来说,producer 将消息投递到 broker 或者直接到 queue 里了,consumer 从 queue 取出消息进行消费,但某些时候由于特定的原因导致 queue 中的某些消息无法被消费,这样的消息如果没有后续的处理,就变成了死信,有死信自然就有了死信队列。
    应用场景:为了保证订单业务的消息数据不丢失,需要使用到 RabbitMQ 的死信队列机制,当消息消费发生异常时,将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

    死信的来源

    消息 TTL 过期
    队列达到最大长度(队列满了,无法再添加数据到 mq 中)
    消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

    死信实战

    代码架构图

    消息 TTL 过期

    生产者代码

    public class Producer {
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     public static void main(String[] argv) throws Exception {
     try (Channel channel = RabbitMqUtils.getChannel()) {
     channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
     //设置消息的 TTL 时间
     AMQP.BasicProperties properties = new 
    AMQP.BasicProperties().builder().expiration("10000").build();
     //该信息是用作演示队列个数限制
     for (int i = 1; i <11 ; i++) {
     String message="info"+i;
     channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, 
    message.getBytes());
     System.out.println("生产者发送消息:"+message);
     }
     }
     }
    }
    

    消费者 C1 代码(启动之后关闭该消费者 模拟其接收不到消息)

    public class Consumer01 {
     //普通交换机名称
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     //死信交换机名称
     private static final String DEAD_EXCHANGE = "dead_exchange";
     public static void main(String[] argv) throws Exception {
     Channel channel = RabbitUtils.getChannel(); //声明死信和普通交换机 类型为 direct
     channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
     channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
     //声明死信队列
     String deadQueue = "dead-queue";
     channel.queueDeclare(deadQueue, false, false, false, null);
     //死信队列绑定死信交换机与 routingkey
     channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
     //正常队列绑定死信队列信息
     Map<String, Object> params = new HashMap<>();
     //正常队列设置死信交换机 参数 key 是固定值
     params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
     //正常队列设置死信 routing-key 参数 key 是固定值
     params.put("x-dead-letter-routing-key", "lisi");
     
     String normalQueue = "normal-queue";
     channel.queueDeclare(normalQueue, false, false, false, params);
     channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
     System.out.println("等待接收消息.....");
     DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     String message = new String(delivery.getBody(), "UTF-8");
     System.out.println("Consumer01 接收到消息"+message);
     };
     channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
     });
     }
    }
    


    消费者 C2 代码(以上步骤完成后 启动 C2 消费者 它消费死信队列里面的消息)

    public class Consumer02 {
     private static final String DEAD_EXCHANGE = "dead_exchange";
     public static void main(String[] argv) throws Exception { Channel channel = RabbitUtils.getChannel();
     channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
     String deadQueue = "dead-queue";
     channel.queueDeclare(deadQueue, false, false, false, null);
     channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
     System.out.println("等待接收死信队列消息.....");
     DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     String message = new String(delivery.getBody(), "UTF-8");
     System.out.println("Consumer02 接收死信队列的消息" + message);
     };
     channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
     });
     }
    }
    

    队列达到最大长度

    1. 消息生产者代码去掉 TTL 属性
    public class Producer {
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     public static void main(String[] argv) throws Exception {
     try (Channel channel = RabbitMqUtils.getChannel()) {
     channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
     //该信息是用作演示队列个数限制
     for (int i = 1; i <11 ; i++) {
     String message="info"+i;
     channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());
     System.out.println("生产者发送消息:"+message);
     }
     }
     }
    }
    
    1. C1 消费者修改以下代码(启动之后关闭该消费者 模拟其接收不到消息)

      注意此时需要把原先队列删除 因为参数改变了
    2. C2 消费者代码不变(启动 C2 消费者)

    消息被拒

    1. 消息生产者代码同上生产者一致
    2. C1 消费者代码(启动之后关闭该消费者 模拟其接收不到消息)
    public class Consumer01 {
     //普通交换机名称
     private static final String NORMAL_EXCHANGE = "normal_exchange";
     //死信交换机名称
     private static final String DEAD_EXCHANGE = "dead_exchange";
     public static void main(String[] argv) throws Exception {
     Channel channel = RabbitUtils.getChannel();
     //声明死信和普通交换机 类型为 direct
     channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
     channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
     //声明死信队列
     String deadQueue = "dead-queue";
     channel.queueDeclare(deadQueue, false, false, false, null);
     //死信队列绑定死信交换机与 routingkey
     channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi"); //正常队列绑定死信队列信息
     Map<String, Object> params = new HashMap<>();
     //正常队列设置死信交换机 参数 key 是固定值
     params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
     //正常队列设置死信 routing-key 参数 key 是固定值
     params.put("x-dead-letter-routing-key", "lisi");
     String normalQueue = "normal-queue";
     channel.queueDeclare(normalQueue, false, false, false, params);
     channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
     System.out.println("等待接收消息.....");
     DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     String message = new String(delivery.getBody(), "UTF-8");
     if(message.equals("info5")){
     System.out.println("Consumer01 接收到消息" + message + "并拒绝签收该消息");
     //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
     channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
     }else {
     System.out.println("Consumer01 接收到消息"+message);
     channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
     }
     };
     boolean autoAck = false;
     channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> {
     });
     }
    }
    

    1. C2 消费者代码不变
      启动消费者 1 然后再启动消费者 2

    延迟队列

    延迟队列概念

    延时队列,队列内部是有序的,最重要的特性就体现在它的延时属性上,延时队列中的元素是希望在指定时间到了以后或之前取出和处理,简单来说,延时队列就是用来存放需要在指定时间被处理的元素的队列。

    延迟队列使用场景

    1. 订单在十分钟之内未支付则自动取消
    2. 新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒。
    3. 用户注册成功后,如果三天内没有登陆则进行短信提醒。
    4. 用户发起退款,如果三天内没有得到处理则通知相关运营人员。
    5. 预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议

    这些场景都有一个特点,需要在某个事件发生之后或者之前的指定时间点完成某一项任务,如:发生订单生成事件,在十分钟之后检查该订单支付状态,然后将未支付的订单进行关闭;看起来似乎使用定时任务,一直轮询数据,每秒查一次,取出需要被处理的数据,然后处理不就完事了吗?如果数据量比较少,确实可以这样做,比如:对于“如果账单一周内未支付则进行自动结算”这样的需求,如果对于时间不是严格限制,而是宽松意义上的一周,那么每天晚上跑个定时任务检查一下所有未支付的账单,确实也是一个可行的方案。但对于数据量比较大,并且时效性较强的场景,如:“订单十分钟内未支付则关闭“,短期内未支付的订单数据可能会有很多,活动期间甚至会达到百万甚至千万级别,对这么庞大的数据量仍旧使用轮询的方式显然是不可取的,很可能在一秒内无法完成所有订单的检查,同时会给数据库带来很大压力,无法满足业务要求而且性能低下。

    RabbitMQ 中的 TTL

    TTL 是什么呢?TTL 是 RabbitMQ 中一个消息或者队列的属性,表明一条消息或者该队列中的所有消息的最大存活时间,单位是毫秒。换句话说,如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列,那么这条消息如果在 TTL 设置的时间内没有被消费,则会成为"死信"。如果同时配置了队列的 TTL 和消息的TTL,那么较小的那个值将会被使用,有两种方式设置 TTL。

    消息设置 TTL

    另一种方式便是针对每条消息设置 TTL

    队列设置 TTL

    第一种是在创建队列的时候设置队列的“x-message-ttl”属性

    两者的区别

    如果设置了队列的 TTL 属性,那么一旦消息过期,就会被队列丢弃(如果配置了死信队列被丢到死信队列中),而第二种方式,消息即使过期,也不一定会被马上丢弃,因为消息是否过期是在即将投递到消费者
    之前判定的,如果当前队列有严重的消息积压情况,则已过期的消息也许还能存活较长时间;另外,还需要注意的一点是,如果不设置 TTL,表示消息永远不会过期,如果将 TTL 设置为 0,则表示除非此时可以直接投递该消息到消费者,否则该消息将会被丢弃。
    前一小节我们介绍了死信队列,刚刚又介绍了 TTL,至此利用 RabbitMQ 实现延时队列的两大要素已经集齐,接下来只需要将它们进行融合,再加入一点点调味料,延时队列就可以新鲜出炉了。想想看,延时队列,不就是想要消息延迟多久被处理吗,TTL 则刚好能让消息在延迟多久之后成为死信,另一面,成为死信的消息都会被投递到死信队列里,这样只需要消费者一直消费死信队列里的消息就完事了,因为里面的消息都是希望被立即处理的消息。

    整合 springboot

    添加依赖

    <dependencies>
     <!--RabbitMQ 依赖-->
     <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>
     <dependency>
     <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId>
     </dependency>
     <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-test</artifactId>
     <scope>test</scope>
     </dependency>
     <dependency>
     <groupId>com.alibaba</groupId>
     <artifactId>fastjson</artifactId>
     <version>1.2.47</version>
     </dependency>
     <dependency>
     <groupId>org.projectlombok</groupId>
     <artifactId>lombok</artifactId>
     </dependency>
     <!--swagger-->
     <dependency>
     <groupId>io.springfox</groupId>
     <artifactId>springfox-swagger2</artifactId>
     <version>2.9.2</version>
     </dependency>
     <dependency>
     <groupId>io.springfox</groupId>
     <artifactId>springfox-swagger-ui</artifactId>
     <version>2.9.2</version>
     </dependency>
     <!--RabbitMQ 测试依赖-->
     <dependency>
     <groupId>org.springframework.amqp</groupId>
     <artifactId>spring-rabbit-test</artifactId>
     <scope>test</scope>
     </dependency>
    </dependencies>
    

    修改配置文件

    spring.rabbitmq.host=182.92.234.71
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123
    

    添加 Swagger 配置类

    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import springfox.documentation.builders.ApiInfoBuilder;
    import springfox.documentation.service.ApiInfo;
    import springfox.documentation.service.Contact;
    import springfox.documentation.spi.DocumentationType;
    import springfox.documentation.spring.web.plugins.Docket;
    import springfox.documentation.swagger2.annotations.EnableSwagger2;
    @Configuration
    @EnableSwagger2
    public class SwaggerConfig {
     @Bean
     public Docket webApiConfig(){
     return new Docket(DocumentationType.SWAGGER_2)
     .groupName("webApi")
     .apiInfo(webApiInfo())
     .select() .build();
     }
     private ApiInfo webApiInfo(){
     return new ApiInfoBuilder()
     .title("rabbitmq 接口文档")
     .description("本文档描述了 rabbitmq 微服务接口定义")
     .version("1.0")
     .contact(new Contact("enjoy6288", "http://atguigu.com", 
    "1551388580@qq.com"))
     .build();
     }
    }
    

    队列 TTL

    代码架构图

    创建两个队列 QA 和 QB,两者队列 TTL 分别设置为 10S 和 40S,然后在创建一个交换机 X 和死信交换机 Y,它们的类型都是 direct,创建一个死信队列 QD,它们的绑定关系如下:

    配置文件类代码

    @Configuration
    public class TtlQueueConfig {
     public static final String X_EXCHANGE = "X";
     public static final String QUEUE_A = "QA";
     public static final String QUEUE_B = "QB";
     public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
     public static final String DEAD_LETTER_QUEUE = "QD";
     // 声明 xExchange
     @Bean("xExchange")
     public DirectExchange xExchange(){
     return new DirectExchange(X_EXCHANGE);
     }
     // 声明 xExchange
     @Bean("yExchange")
     public DirectExchange yExchange(){
     return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
     }
     //声明队列 A ttl 为 10s 并绑定到对应的死信交换机 @Bean("queueA")
     public Queue queueA(){
     Map<String, Object> args = new HashMap<>(3);
     //声明当前队列绑定的死信交换机
     args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
     //声明当前队列的死信路由 key
     args.put("x-dead-letter-routing-key", "YD");
     //声明队列的 TTL
     args.put("x-message-ttl", 10000);
     return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
     }
     // 声明队列 A 绑定 X 交换机
     @Bean
     public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,
     @Qualifier("xExchange") DirectExchange xExchange){
     return BindingBuilder.bind(queueA).to(xExchange).with("XA");
     }
     //声明队列 B ttl 为 40s 并绑定到对应的死信交换机
     @Bean("queueB")
     public Queue queueB(){
     Map<String, Object> args = new HashMap<>(3);
     //声明当前队列绑定的死信交换机
     args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
     //声明当前队列的死信路由 key
     args.put("x-dead-letter-routing-key", "YD");
     //声明队列的 TTL
     args.put("x-message-ttl", 40000);
     return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
     }
     //声明队列 B 绑定 X 交换机
     @Bean
     public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
     @Qualifier("xExchange") DirectExchange xExchange){
     return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
     }
     //声明死信队列 QD
     @Bean("queueD")
     public Queue queueD(){
     return new Queue(DEAD_LETTER_QUEUE);
     }
     //声明死信队列 QD 绑定关系
     @Bean
     public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
     @Qualifier("yExchange") DirectExchange yExchange){
     return BindingBuilder.bind(queueD).to(yExchange).with("YD");
     }
    }
    

    消息生产者代码

    @Slf4j
    @RequestMapping("ttl")
    @RestController
    public class SendMsgController {
     @Autowired
     private RabbitTemplate rabbitTemplate;
     @GetMapping("sendMsg/{message}")
     public void sendMsg(@PathVariable String message){ log.info("当前时间:{},发送一条信息给两个 TTL 队列:{}", new Date(), message);
     rabbitTemplate.convertAndSend("X", "XA", "消息来自 ttl 为 10S 的队列: "+message);
     rabbitTemplate.convertAndSend("X", "XB", "消息来自 ttl 为 40S 的队列: "+message);
     } 
    }
    

    消息消费者代码

    @Slf4j
    @Component
    public class DeadLetterQueueConsumer {
     @RabbitListener(queues = "QD")
     public void receiveD(Message message, Channel channel) throws IOException {
     String msg = new String(message.getBody());
     log.info("当前时间:{},收到死信队列信息{}", new Date().toString(), msg);
     }
    }
    

    发起一个请求 http://localhost:8080/ttl/sendMsg/嘻嘻嘻

    第一条消息在 10S 后变成了死信消息,然后被消费者消费掉,第二条消息在 40S 之后变成了死信消息,然后被消费掉,这样一个延时队列就打造完成了。
    不过,如果这样使用的话,岂不是每增加一个新的时间需求,就要新增一个队列,这里只有 10S 和 40S两个时间选项,如果需要一个小时后处理,那么就需要增加 TTL 为一个小时的队列,如果是预定会议室然后提前通知这样的场景,岂不是要增加无数个队列才能满足需求?

    延时队列优化

    代码架构图

    在这里新增了一个队列 QC,绑定关系如下,该队列不设置 TTL 时间

    配置文件类代码

    @Component
    public class MsgTtlQueueConfig {
     public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
     public static final String QUEUE_C = "QC";
     //声明队列 C 死信交换机
     @Bean("queueC")
     public Queue queueB(){
     Map<String, Object> args = new HashMap<>(3);
     //声明当前队列绑定的死信交换机
     args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
     //声明当前队列的死信路由 key
     args.put("x-dead-letter-routing-key", "YD");
     //没有声明 TTL 属性
     return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
     }
     //声明队列 B 绑定 X 交换机
     @Bean
     public Binding queuecBindingX(@Qualifier("queueC") Queue queueC,
     @Qualifier("xExchange") DirectExchange xExchange){
     return BindingBuilder.bind(queueC).to(xExchange).with("XC");
     }
    }
    

    消息生产者代码

    @GetMapping("sendExpirationMsg/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message,@PathVariable String ttlTime) {
     rabbitTemplate.convertAndSend("X", "XC", message, correlationData ->{
     correlationData.getMessageProperties().setExpiration(ttlTime);
     return correlationData;
     });
     log.info("当前时间:{},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(),ttlTime, message);
    }
    

    发起请求
    http://localhost:8080/ttl/sendExpirationMsg/你好 1/20000
    http://localhost:8080/ttl/sendExpirationMsg/你好 2/2000

    看起来似乎没什么问题,但是在最开始的时候,就介绍过如果使用在消息属性上设置 TTL 的方式,消息可能并不会按时“死亡“,因为 RabbitMQ 只会检查第一个消息是否过期,如果过期则丢到死信队列,如果第一个消息的延时时长很长,而第二个消息的延时时长很短,第二个消息并不会优先得到执行。

    Rabbitmq 插件实现延迟队列

    上文中提到的问题,确实是一个问题,如果不能实现在消息粒度上的 TTL,并使其在设置的 TTL 时间及时死亡,就无法设计成一个通用的延时队列。那如何解决呢,接下来我们就去解决该问题。

    安装延时队列插件

    在官网上下载 https://www.rabbitmq.com/community-plugins.html,下载
    rabbitmq_delayed_message_exchange 插件,然后解压放置到 RabbitMQ 的插件目录。
    进入 RabbitMQ 的安装目录下的 plgins 目录,执行下面命令让该插件生效,然后重启 RabbitMQ
    /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
    rabbitmq-plugins enable rabbitmq_delayed_message_exchange

    代码架构图

    在这里新增了一个队列 delayed.queue,一个自定义交换机 delayed.exchange,绑定关系如下:

    配置文件类代码

    在我们自定义的交换机中,这是一种新的交换类型,该类型消息支持延迟投递机制 消息传递后并不会立即投递到目标队列中,而是存储在 mnesia(一个分布式数据系统)表中,当达到投递时间时,才投递到目标队列中。

    @Configuration
    public class DelayedQueueConfig {
     public static final String DELAYED_QUEUE_NAME = "delayed.queue";
     public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
     public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
     @Bean
     public Queue delayedQueue() {
     return new Queue(DELAYED_QUEUE_NAME);
     }
     //自定义交换机 我们在这里定义的是一个延迟交换机
     @Bean
     public CustomExchange delayedExchange() {
     Map<String, Object> args = new HashMap<>();
     //自定义交换机的类型
     args.put("x-delayed-type", "direct");
     return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, 
    args);
     }
     @Bean
     public Binding bindingDelayedQueue(@Qualifier("delayedQueue") Queue queue,
     @Qualifier("delayedExchange") CustomExchange 
    delayedExchange) {
     return 
    BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
     }
    }
    

    消息生产者代码

    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
    @GetMapping("sendDelayMsg/{message}/{delayTime}")
    public void sendMsg(@PathVariable String message,@PathVariable Integer delayTime) {
     rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message, 
    correlationData ->{
     correlationData.getMessageProperties().setDelay(delayTime);
     return correlationData; });
     log.info(" 当 前 时 间 : {}, 发送一条延迟 {} 毫秒的信息给队列 delayed.queue:{}", new 
    Date(),delayTime, message);
    }
    

    消息消费者代码

    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    @RabbitListener(queues = DELAYED_QUEUE_NAME)
    public void receiveDelayedQueue(Message message){
     String msg = new String(message.getBody());
     log.info("当前时间:{},收到延时队列的消息:{}", new Date().toString(), msg);
    }
    

    发起请求:
    http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000
    http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000

    第二个消息被先消费掉了,符合预期

    总结

    延时队列在需要延时处理的场景下非常有用,使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性,如:消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外,通过 RabbitMQ 集群的特性,可以很好的解决单点故障问题,不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
    当然,延时队列还有很多其它选择,比如利用 Java 的 DelayQueue,利用 Redis 的 zset,利用 Quartz或者利用 kafka 的时间轮,这些方式各有特点,看需要适用的场景

    展开全文
  • RabbitMQ 消息队列

    千次阅读 2021-11-08 16:24:45
    使用较多的消息队列产品:RabbitMQ,RocketMQ,ActiveMQ,ZeroMQ,Kafka 等。 异步处理 用户注册后,需要发送验证邮箱和手机验证码。 将注册信息写入数据库,发送验证邮件,发送手机,三个步骤全部完成后,返回给...
  • 指把要传输的数据(消息)放在队列中,用队列机制来实现消息传递——生产者产生消息并把消息放入队列,然后由消费者去处理。消费者可以到指定队列拉取消息,或者订阅相应的队列,由MQ服务端给其推送消息。(来源:...
  • 本章主要目的是为了后续spring-cloud-bus做准备,讲述在Linux Centos7操作系统中搭建 RabbitMQ…- 什么是RabbitMQRabbitMQ 是一个使用 Erlang 语言实现的一个高并发高可靠AMQP消息队列服务器,并且支持 多种消息协议...
  • app 清除的命令为: rabbitmqctl reset 重新启动命令为: rabbitmqctl start_app ps 查看所有队列命令: rabbitmqctl list_queues 注意 命令行批量删除相当于重置RabbitMQ,请谨慎操作,配置和队列都会被清除。...
  • RabbitMq——延迟队列

    千次阅读 2022-03-18 20:33:09
    延迟队列:用于存放需要在指定时间后(TTL)被执行的消息的队列,延迟队列消息过期后会被存放到死信队列,消费者不断对死信队列进行消费。 TTL:消息存活的最大时间,单位为毫秒,在TTL时间内,若消息未被消费,则...
  • RabbitMQ—消息队列

    千次阅读 2022-01-26 16:09:57
    RabbitMQ是一个在AMQP( 一个提供统一消息服务的应用层标准高级消息队列协议)基础上完成的,可复用的企业消息系统。他遵循Mozilla Public License开源协议。 主要特性: 保证可靠性:使用一些机制来保证可靠性,如...
  • 清空rabbitmq队列数据

    千次阅读 2022-04-12 14:01:26
    建议直接在rabbitmq管理平台操作,找到目标队列,purge一下即可。
  • 1.首先进入rabbitMq的目录 2.清除命令 sudo rabbitmqctl stop_app;sudo rabbitmqctl reset;sudo rabbitmqctl start_app;sudo rabbitmqctl add_user crs crs;sudo rabbitmqctl set_permissions -p / crs ".*" ...
  • 根据名字可以看出 这个队列用来保存 被丢弃或者被删除的信息等 被丢弃被删除的信息跟死了一样 简称死信 把这些数据放到一个队列中 称为死信队列 当一个消息被删除变为Dead message(读音 带的 吗c只 这个单词就是...
  • 1. MQ 消息队列 1.1 MQ的概念 1.2 为什么要用MQ? 1.3 MQ分类 1.4 MQ的选择 2. RabbitMQ 的四大核心概念 3. 各个名词的理解 4. RabbitMQ 安装 4.1 下载erlong安装包和rabbitmq安装包 4.2 安装erlang 、 安装rabbitmq...
  • RabbitMQ清除队列中的所有消息

    千次阅读 2018-09-25 19:25:00
    最近在研究 RabbitMQ 消息队列,安装好进行测试的时候发觉在一个名为 MyRabbitMQ 的消息队列中已经插入了大量的数据。 最后不得不找方法清除。 linux或者windows 输入命令: rabbitmqctl list_queues 查看...
  • 注:【本文为个人学习中所记录,可能包含从网上摘抄的内容,侵删!...关闭应用# 可以先执行 rabbitmqctl list_queues,查看现在的队列信息 rabbitmqctl stop_app 清除rabbitmqctl reset 启动应用rabbi.
  • 1. 业务场景 ... 思路:发一条跟订单相关的消息,30分钟以后被消费,在消费...利用RabbitMQ的死信队列实现 定时任务比较容易实现,比如每隔一分钟扫描一次,查出30分钟之前未付款的订单,把状态改为关闭。但是如果瞬间要
  • 1、使用命令文件操作rabbitMQ cd /root/rabbitmq/rabbitmq_server/sbin/ ./rabbitmq-server -detached 启动rabbit ./rabbitmqctl stop 停止rabbit 2、使用全局变量操作rabbitMQ vim /etc/profile export ...
  • rabbitmq批量删除队列

    2020-12-17 06:22:39
    有些时候,我们需要批量的删除rabbitmq中的队列,尤其是对于那些客户端配置了队列不存在时自动创建,但断开时不自动删除的应用来说。rabbitmqctl并没有包含直接管理队列的功能,其提供的vhost粒度有时候过大以至于...
  • LINUX下安装MQ: 参考: https://blog.csdn.net/tony308001970/article/details/53034320 常用操作: $ sudo chkconfig rabbitmq-server on # 添加开机启动RabbitMQ服务 $ sud...
  • 0X00 安装环境首先是在Linux上安装rabbitmq# 环境为CentOS 7yum install rabbitmq-server # 安装RabbitMQsystemctl start rabbitmq-server # 启动systemctl enable rabbitmq-server # 开机自启systemctl stop ...
  • 文章目录RabbitMQ的相关概念RabbitMQ的几种模式Exchange的四种类型Linux系统上Rabbitmq的安装RabbitMQ的Web界面操作快速体验参考链接 RabbitMQ的相关概念 AMPQ:Advanced Message Queue Protocol,RabbitMQ是erlang...
  • RabbitMQ消息队列(一)

    千次阅读 2021-11-04 12:23:51
    RabbitMQ消息队列中间件 直接安装在Linux上 启动服务 systemctl start rabbitmq-server 查看服务状态 systemctl status rabbitmq-server 停止服务 systemctl stop rabbitmq-server 开机启动服务 systemctl ...
  • RabbitMQ中可以虚拟消息服务器VirtualHost,每个VirtualHost相当月一个相对独立的RabbitMQ服务器,每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 在RabbitMQ中无法通过AMQP创建Virtual...
  • Linux RabbitMQ高可用mirror镜像集群策略配置 -清理单机和网络开发 1.关闭原先的单节点 2.阿里网络安全组开放对应的端口 3.防火墙一定要关闭 -准备3个节点安装好rabbitmq,形成集群 (记得每个节点间隔几十秒再启动...
  • 延迟队列可以通过设置生产者发送设置有ttl时间的信息给队列来实现,但这种方式中的消息可能不会按时成为死信,而且rabbitmq每次只能检测第一个消息是否过期,如果过期就丢入死信队列,第一个消息延迟时间很长,第二...
  • 选择rabbitmq_delayed_message_exchange(一个向RabbitMQ添加延迟消息(或计划消息)的插件)进行下载 现在下载的版本是3.7或者3.8,所以rabbitmq需要3.7以上 下面是rabbitmq_delayed_message_exchange的下载地址: ...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 8,914
精华内容 3,565
关键字:

linux rabbitmq删除队列