- 浏览: 626455 次
- 性别:
- 来自: 北京
文章分类
- 全部博客 (819)
- java开发 (110)
- 数据库 (56)
- javascript (30)
- 生活、哲理 (17)
- jquery (36)
- 杂谈 (15)
- linux (62)
- spring (52)
- kafka (11)
- http协议 (22)
- 架构 (18)
- ZooKeeper (18)
- eclipse (13)
- ngork (2)
- dubbo框架 (6)
- Mybatis (9)
- 缓存 (28)
- maven (20)
- MongoDB (3)
- 设计模式 (3)
- shiro (10)
- taokeeper (1)
- 锁和多线程 (3)
- Tomcat7集群 (12)
- Nginx (34)
- nodejs (1)
- MDC (1)
- Netty (7)
- solr (15)
- JSON (8)
- rabbitmq (32)
- disconf (7)
- PowerDesigne (0)
- Spring Boot (31)
- 日志系统 (6)
- erlang (2)
- Swagger (3)
- 测试工具 (3)
- docker (17)
- ELK (2)
- TCC分布式事务 (2)
- marathon (12)
- phpMyAdmin (12)
- git (3)
- Atomix (1)
- Calico (1)
- Lua (7)
- 泛解析 (2)
- OpenResty (2)
- spring mvc (19)
- 前端 (3)
- spring cloud (15)
- Netflix (1)
- zipkin (3)
- JVM 内存模型 (5)
- websocket (1)
- Eureka (4)
- apollo (2)
- idea (2)
- go (1)
- 业务 (0)
- idea开发工具 (1)
最新评论
-
sichunli_030:
对于频繁调用的话,建议采用连接池机制
配置TOMCAT及httpClient的keepalive以高效利用长连接 -
11想念99不见:
你好,我看不太懂。假如我的项目中会频繁调用rest接口,是要用 ...
配置TOMCAT及httpClient的keepalive以高效利用长连接
rabbitMQ中consumer通过建立到queue的连接,创建channel对象,通过channel通道获取message,
Consumer可以声明式的以API轮询poll的方式主动从queue的获取消息,也可以通过订阅的方式被动的从Queue中消费消息,
最近翻阅了基于java的客户端的相关源码,简单做个分析。
编程模型伪代码如下:
ConnectionFactory factory = new ConnectionFactory();
Connection conn = factory.newConnection();
Channel channel=conn.createChannel();
创建Connection需要指定MQ的物理地址和端口,是socket tcp物理连接,而channel是一个逻辑的概念,支持在tcp连接上创建多个MQ channel
以下是基于channel上的两种消费方式。
1、Subscribe订阅方式
订阅方式其实是向queue注册consumer,通过rpc向queue server发送注册consumer的消息,rabbitMQ Server在收到消息后,根据消息的内容类型判断这是一个订阅消息,
这样当MQ 中queue有消息时,会自动把消息通过该socket(长连接)通道发送出去。
参见ChannelN中的方法
Consumer接收消息的过程:
创建Connection后,会启动MainLoop后台线程,循环从socket(FrameHandler)中获取数据包(Frame),调用channel.handleFrame(Frame frame)处理消息,
每个Consumer都有一个BlockQueue,用于缓存从socket中获取的消息。
接下来,Consumer对象就可以调用api来从客户端缓存的_queue中依次获取消息,进行消费,参见QueueingConsumer.nextDelivery()
对于这种长连接的方式,没看到心跳功能,以防止长连接的因网络等原因连接失效
2、poll API方式
ChannelN:
GetResponse basicGet(String queue, boolean autoAck)
这种方式比较简单,直接通过RPC从MQ Server端获取队列中的消息
参考:http://blog.csdn.net/liyantianmin/article/details/46696029
Consumer可以声明式的以API轮询poll的方式主动从queue的获取消息,也可以通过订阅的方式被动的从Queue中消费消息,
最近翻阅了基于java的客户端的相关源码,简单做个分析。
编程模型伪代码如下:
ConnectionFactory factory = new ConnectionFactory();
Connection conn = factory.newConnection();
Channel channel=conn.createChannel();
创建Connection需要指定MQ的物理地址和端口,是socket tcp物理连接,而channel是一个逻辑的概念,支持在tcp连接上创建多个MQ channel
以下是基于channel上的两种消费方式。
1、Subscribe订阅方式
boolean autoAck = false; channel.basicConsume(queueName, autoAck, "myConsumerTag", new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String routingKey = envelope.getRoutingKey(); String contentType = properties.contentType; long deliveryTag = envelope.getDeliveryTag(); // (process the message components here ...) channel.basicAck(deliveryTag, false); } });
订阅方式其实是向queue注册consumer,通过rpc向queue server发送注册consumer的消息,rabbitMQ Server在收到消息后,根据消息的内容类型判断这是一个订阅消息,
这样当MQ 中queue有消息时,会自动把消息通过该socket(长连接)通道发送出去。
参见ChannelN中的方法
public String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, final Consumer callback) throws IOException { ...... rpc((Method) new Basic.Consume.Builder() .queue(queue) .consumerTag(consumerTag) .noLocal(noLocal) .noAck(autoAck) .exclusive(exclusive) .arguments(arguments) .build(), k); try { return k.getReply(); } catch(ShutdownSignalException ex) { throw wrap(ex); } }
Consumer接收消息的过程:
创建Connection后,会启动MainLoop后台线程,循环从socket(FrameHandler)中获取数据包(Frame),调用channel.handleFrame(Frame frame)处理消息,
public void handleFrame(Frame frame) throws IOException { AMQCommand command = _command; if (command.handleFrame(frame)) { // 对消息进行协议assemble _command = new AMQCommand(); // prepare for the next one handleCompleteInboundCommand(command);//对消息消费处理 } } ChannelN.handleCompleteInboundCommand ---ChannelN.processAsync ----dispatcher.handleDelivery ---QueueingConsumer.handleDelivery ---this._queue.add(new Delivery(envelope, properties, body));//消息最终放到队列中
每个Consumer都有一个BlockQueue,用于缓存从socket中获取的消息。
接下来,Consumer对象就可以调用api来从客户端缓存的_queue中依次获取消息,进行消费,参见QueueingConsumer.nextDelivery()
对于这种长连接的方式,没看到心跳功能,以防止长连接的因网络等原因连接失效
2、poll API方式
ChannelN:
GetResponse basicGet(String queue, boolean autoAck)
这种方式比较简单,直接通过RPC从MQ Server端获取队列中的消息
参考:http://blog.csdn.net/liyantianmin/article/details/46696029
发表评论
-
RocketMQ教程,包含所有MQ核心知识点!
2022-04-28 13:49 156RocketMQ教程,包含所有MQ核心知识点 原创 | Ja ... -
rabbitmq死信队列和延时队列的使用
2021-12-25 23:19 241rabbitmq死信队列和延时队列的使用 -
IM消息送达保证机制实现(一):保证在线实时消息的可靠投递
2021-12-14 11:49 165[url=http://www.52im.net/thread ... -
RabbitMQ高级特性TTL队列/消息
2021-09-04 22:47 211RabbitMQ高级特性-TTL队列/消息 RabbitMQ ... -
如何保证消息不丢失,消息顺序执行-面试
2021-05-26 20:24 231关于MQ的几件小事(四)如何保证消息不丢失 如何保证Rab ... -
RabbitMQ 相关问题汇总
2017-06-28 17:43 427RabbitMQ 相关问题汇总 rabbitmq基础概念与基 ... -
rabbitMq集成Spring后,消费者设置手动ack,并且在业务上控制是否ack
2017-06-12 20:30 2156http://blog.csdn.net/u010841296 ... -
rabbitmq——镜像队列
2016-12-02 20:05 11191. 镜像队列的设置 镜像队列的配置通过添加policy完成 ... -
RabbitMQ 内部实现
2016-12-01 14:41 1014http://blog.csdn.net/joeyon1985 ... -
OpenStack RabbitMQ 集群-后续整理
2016-12-01 14:18 502参考:http://www.iyunv.com/thread- ... -
RabbitMQ (三) 发布/订阅
2016-11-30 19:53 5561、转发器(Exchanges) ... -
RabbitMQ学习(六)之远程过程调用(RPC)
2016-11-30 14:31 835在一般使用RabbitMQ做RPC很容易。客户端发送一个请求消 ... -
RabbitMQ学习之Headers交换类型
2016-11-28 10:51 791Headers类型的exchange使用的比较少,它也是忽略r ... -
RabbitMQ能打开的最大连接数
2016-11-28 10:29 2572转自:http://blog.csdn.net/huoyuns ... -
RabbitMQ基础知识
2016-11-28 10:25 516Routing key由生产者指定。Binding key由消 ... -
RabbitMQ Consumer获取消息的两种方式(poll,subscribe)解析
2016-11-25 21:36 884rabbitMQ中consumer通过建立到queue的连接, ... -
解决RabbitMQ远程不能访问的问题
2016-11-24 15:18 1163刚刚安装的RabbitMQ-Server-3.3.5,并且 ... -
RabbitMQ用户角色及权限控制
2016-11-24 11:08 1751RabbitMQ:基本命令 rabbitmq的安装、启动和停 ... -
publish消息确认
2016-11-23 18:01 664Using standard AMQP, the only w ... -
rabbitMQ ConfirmListener
2016-11-23 15:53 2153消息消费者 操作步骤: 1. 创建连接工厂Connection ...
相关推荐
为了确保手动投资的优先级高于自动投资,我们可以创建两个不同的消息队列,分别对应这两种投资类型。例如,可以命名为"manual_investment_queue"和"auto_investment_queue"。 1. **手动投资流程**: - 生产者在...
消费者(ConsoleMQConsumer)负责接收和处理消息,而生产者(ConsoleMQPublish)则负责发送消息到RabbitMQ服务器。在实现过程中,你需要创建连接工厂、通道(Channel),然后定义交换机和队列,最后进行绑定。此外,...
3. **消息确认机制**:为了确保消息不丢失,RabbitMQ提供了消息确认机制,包括手动确认(Manual Acknowledgement)和自动确认(Automatic Acknowledgement)两种方式。 4. **集群部署**:通过集群部署可以实现...
在技术方案上,有两种主要的实现方式: 1. 针对Queue设置延迟时间。这是通过配置TTL(Time To Live)属性实现的,即设置队列消息过期时间。这种方式的优点是维护简单、客户端完全透明;缺点是发送方无法自定义延迟...
这个名为"SpringBoot-RabbitMQ生产者和消费者.7z"的压缩包包含两个关键部分:`rabbitmq-consumer`(消费者)和`rabbitmq-provider`(生产者),它们演示了如何在SpringBoot应用中集成并利用RabbitMQ进行数据交换。...
在本资源中,我们有两个安装包:RabbitMQ 3.12.10和Erlang 26.0.2,这两个是RabbitMQ运行所必需的组件。 Erlang是一种函数式编程语言,以其并发能力、容错性和热代码升级功能而闻名。在RabbitMQ中,Erlang作为基础...
1. **确认机制**:RabbitMQ支持两种确认模式——自动确认和手动确认。自动确认模式下,一旦RabbitMQ将消息从队列中取出并传递给消费者,就认为消息已被成功处理。但在复杂场景下,可能需要手动确认,即消费者接收到...
路由键和绑定键可以包含点号(".")分隔的单词,通配符有两种:“*”(匹配一个单词)和“#”(匹配零个或多个单词)。这允许消费者订阅多个相关的路由键,提高灵活性。 3. **Fanout模式**: Fanout模式是广播模式...
RabbitMQ是一种广泛应用的消息中间件,它基于AMQP(Advanced Message Queuing Protocol)协议,提供可靠的消息传递服务。在分布式系统中,RabbitMQ扮演着数据交换中心的角色,使得不同服务之间可以通过异步通信方式...
1. **消息确认**:在RabbitMQ中,消息确认(Message Acknowledgement)是一种确保消息被正确处理的机制。当消费者接收到消息后,它需要发送一个确认信号给RabbitMQ,表明消息已被处理。如果RabbitMQ没有收到确认,它...
### RabbitMQ消费端幂等性概念及解决方案 #### 一、幂等性的基本概念 **幂等性**是指用户对于同一个操作发起一次请求或多次请求的结果保持一致的特性。例如,在数据库操作中,通过实现乐观锁机制来确保幂等性。...
RabbitMQ提供了一种可靠、可扩展和灵活的方式来在分布式系统中传递消息。在本项目中,RabbitMQ作为消息中间件,负责接收生产者发送的消息,并将这些消息推送到消费者端。它确保了消息的可靠传输,即使在服务端或...
JMS则是一个Java平台的消息服务API,它定义了一组统一的消息操作接口,要求使用Java语言进行消息的生产和消费,规定了两种消息模式。 综上所述,消息队列是一个重要的中间件技术,尤其在分布式系统和异步处理系统中...
在RabbitMQ中,有两种方式实现消息持久化: - **Exchange持久化**:交换机的配置信息会被保存在磁盘上,即使RabbitMQ服务器重启,交换机依然存在,新的消息可以根据之前的配置继续路由。 - **Queue持久化**:队列...
路由键可以包含单词(由`.`分隔),并且支持两种通配符:“*”(匹配一个单词)和“#”(匹配零个或多个单词)。例如,队列`news.europe`可以绑定到`news.#`,接收所有新闻类别的欧洲消息,而`news.sports.europe`...
测试是在RabbitMQ 3.6.2版本上进行的,采用了两种不同的部署模式:单机模式和集群模式。对于单机模式,分配了2核4GB的资源;而在集群模式下,每个节点配置为4核8GB,优先级较高,以确保更佳的性能表现。 2.2 **性能...
在IT行业中,消息队列(Message Queue)是一种重要的中间件技术,它允许不同的系统之间进行异步通信,提高系统的解耦性和可扩展性。RabbitMQ作为一款开源的消息队列服务器,广泛应用于微服务架构中。Spring框架是...
- **消息确认**:RabbitMQ支持两种确认模式——publisher confirms和consumer acknowledges,确保消息的可靠投递。 - **消息持久化**:通过设置消息和队列的持久化属性,即使RabbitMQ重启,消息也不会丢失。 6. *...
- topic:基于通配符的路由,支持“.”和“*”两种模式。 - headers:基于消息头的匹配,不使用路由键。 - rabbitmq_exchange_type:自定义类型的交换机,可以实现特定逻辑。 8. **RabbitMQ的死信队列(Dead ...
消息队列MQ,全称Message Queue,是一种在应用程序之间用于通信的技术,主要功能是作为生产者与消费者之间的缓冲。在消息队列模型中,生产者不断地向队列中发布消息,而消费者则从队列中获取并处理这些消息。这种...