AMQP-RabbitMQ/2/工作队列
2. 工作队列 Work queues
Distributing tasks among workers
消息将发送给c1或者c2
个人理解
- 生产者定义Queue,并向该队列发送消息
- 多个消费者可以从指定的同一个Queue中读取消息。每条消息只会发送给其中某一个消费者。
- 生产者
package com.futao.springmvcdemo.mq.rabbit.workqueue; import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools; import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import lombok.Cleanup; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; /** * 简单发送者 * * @author futao * Created on 2019-04-22. */ @Slf4j public class Send { @SneakyThrows public static void main(String[] args) { @Cleanup Connection connection = RabbitMqConnectionTools.getConnection(); @Cleanup Channel channel = connection.createChannel(); //开启持久化消息 boolean durable = true; //定义一个队列 channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null); String msg = "Hello RabbitMq!"; for (int i = 0; i < 20; i++) { channel.basicPublish("", RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), null, (msg + i).getBytes()); log.info("Send msg:[{}] success", (msg + i)); } } }
- 消费者1 - 每1秒处理一条
package com.futao.springmvcdemo.mq.rabbit.workqueue; import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools; import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; /** * 简单消费者 * * @author futao * Created on 2019-04-22. */ @Slf4j public class RecvOne { @SneakyThrows public static void main(String[] args) { Channel channel = RabbitMqConnectionTools.getChannel(); //开启持久化消息 boolean durable = true; channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null); log.info("Waiting for message..."); DeliverCallback deliverCallback = ((consumerTag, message) -> { log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag); //acknowledgment应答 channel.basicAck(message.getEnvelope().getDeliveryTag(), false); try { Thread.sleep(1000); } catch (Exception e) { } }); //关闭自动应答 boolean autoAck = false; channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> { }); } }
- 消费者2 - 每2秒处理一条
package com.futao.springmvcdemo.mq.rabbit.workqueue; import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools; import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; /** * 简单消费者 * * @author futao * Created on 2019-04-22. */ @Slf4j public class RecvTwo { @SneakyThrows public static void main(String[] args) { Channel channel = RabbitMqConnectionTools.getChannel(); //开启持久化消息 boolean durable = true; channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null); log.info("Waiting for message..."); DeliverCallback deliverCallback = ((consumerTag, message) -> { log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag); //acknowledgment应答 channel.basicAck(message.getEnvelope().getDeliveryTag(), false); try { Thread.sleep(2000); } catch (Exception e) { } }); //关闭自动应答 boolean autoAck = false; channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> { }); } }
-
结果
- 生产者
- 消费者1日志
- 消费者2日志
- 生产者
- 特点: 多个消费者之间,不论消息的处理速度,都是平均分发(公平分发)。你一个,我一个,他一个。此时是公平队列
-
注意:
- 定义队列的时候,设置是否开启消息的持久化(该设置需要同时在生产者和消费者设置)
//开启持久化消息 boolean durable = true; channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
- 如果消息队列已经存在,则不可以修改相应的配置,必须删除原有的队列,或者新建一个新的队列。
- 关闭自动应答(开启手动应答),可以防止消息在未被正确消费的情况下被Rabbitmq从队列内存中删除。
实现工作队列下的非公平队列
消费者设置一次只发送一条消息,并且在被正确消费之前发继续发送下一条消息。从而使得消费快的消费者比消费慢的消费者消费更多的消息
//告诉rabbitmq一次只发送一条消息,并且在前一个消息未被处理或者消费之前,不继续发送下一个消息 channel.basicQos(1);
- 测试结果
此时明显打破了消息的公平分发,消费快的消费者接收到的消息更多。
如果有两个消费者,其中一个ConsumerA设置了Qos=1
,另一个ConsumerB没有设置。经过我的测试,ConsumerA会获得大量的消息,都积压在ConsumerA,而ConsumerB获得消息很少。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
AMQP-RabbitMQ/1/概念/一对一简单模型
JMS,AMQP,MQTT的区别与联系 JMS Java消息传递服务(Java Messaging Service ) AMQP 高级消息队列协议(Advanced Message Queueing Protocol ) MQTT消息队列遥测传输(Message Queueing Telemetry Transport ) 简单理解:*JMS是专门为Java设计的一套消息服务API,像ActiveMQ就是对它的实现*AMQP为了解决不同平台之间的通信问题,定义了一种名为amqp的通信协议,从而实现平台和语言无关性。*MQTT也是一种通信协议。相比于AMQP的复杂性,它简单的多。所以amqp用于处理相对较重的任务,如两个系统平台之间的消息传输。而mqtt因为非常轻量,所以大量应用于物联网。 RabbitMQ常用命令 修改密码 rabbitmqctl change_password Username Newpassword 显示所有用户 rabbitmqctl list_users 启动 rabbitmq-server start 关闭 rabbitmqctl stop homebrew安...
- 下一篇
Linux零碎笔记, 及一些命令
001.gui与tui gui, 完全的图形界面. tui, 终端下的图形界面. 002.补全命令 输入前缀, 按一次Tab. 若前缀不唯一, 按2次Tab, 出现相关命令列表. 003.RHEL7.2实现上网功能 1.固定的可用联网IP地址 2.关闭防火墙 3.关闭selinux 4.光盘自动挂载 5.配置好本地yum源 004.linux系统安装分区(标准分区) 1.boot分区:引导分区(启动系统用的), grub(引导器), 内核文件 200M 2.swap分区:内存扩展分区, 一般最多8G或16G 1000M 3./ 根:所有文件的根 10G 4.sda:linux下的第一个磁盘 5.RHEL7使用xfs文件系统, RHEL6使用ext4文件系统, Win7/10使用NTFS文件系统 005.手动配置以太网_IPv4设置 1.ip地址:192.168.0.71(网段192.168一样, 且是没用过的ip) 2.子网掩码:24 3.网关:192.168.0.1 4.DNS服务器:8.8.8.8(或114.114.114.114) 006.ubuntu第一次切换到root用户失败...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- SpringBoot2整合Redis,开启缓存,提高访问速度
- SpringBoot2整合Thymeleaf,官方推荐html解决方案
- CentOS8安装MyCat,轻松搞定数据库的读写分离、垂直分库、水平分库
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- Red5直播服务器,属于Java语言的直播服务器
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- Docker安装Oracle12C,快速搭建Oracle学习环境
- Hadoop3单机部署,实现最简伪集群