Category Archives: Message Queue

Kafka 内存池MemoryPool 设计

GarbageCollectedMemoryPool 是一个非常巧妙的调试工具。它通过继承 SimpleMemoryPool 复用了内存额度控制的逻辑，然后通过重写钩子方法和结合Java的引用机制，增加了一层“内存泄漏”的监控。

正常流程：tryAllocate() -> bufferToBeReturned (登记) -> 使用 -> release() -> bufferToBeReleased (注销) -> ByteBuffer 失去引用 -> 被GC。
泄漏流程：tryAllocate() -> bufferToBeReturned (登记) -> 使用 -> 忘记调用 release() -> ByteBuffer 失去引用 -> 被GC -> BufferReference 进入队列 -> 后台线程检测到，并从 buffersInFlight 中找到了登记信息 -> 报告错误。

RabbitMQ基本实践

声明：本文是《RabbitMQ实战指南》（朱忠华 著）学习笔记，仅供本人学习研究之用，如若喜欢请购买正版书籍。如有侵权，请联系删除。

用docker部署RabbitMQ环境

前置条件： 已经安装好docker 1.查找镜像（有2种方式） ①登录rabbitmq官网找到docker镜像，选择想要的镜像的tag https://www.rabbitmq.com/download.html https://hub.docker.com/_/rabbitmq 如果需要访问web管理页面，就选择tag为management的 ps：带有alpine的是用最小linux镜像构建的，体积最小可以达5M初学者不建议这么折腾，而且 Alpine Linux使用了muslmusl实现的DNS服务不会使用resolv.conf文件中的search和domain两个配置，通过DNS来进行服务发现时需要注意。,带有-management的是带有web控制台   ②直接用docker search 搜索，默认下载标签为latest的镜像（无法打开web管理页面）

ps：如果是普通用户登录，需要sudo，不然会提示错误

2.下载镜像（有时候网络问题超时，多尝试几次即可。我这里选择的是可以访问web管理界面的tag）

3.创建容器并运行（15672是管理界面的端口，5672是服务的端口。这里顺便将管理系统的用户名和密码设置为admin admin）

  from:https://www.cnblogs.com/yy-cola/p/11089800.html

Kafka、RocketMQ、RabbitMQ的优劣势比较

消息队列选择建议
1.Kafka

Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输，适合产生大量数据的互联网服务的数据收集业务。

大型公司建议可以选用，如果有日志采集功能，肯定是首选kafka了。

2.RocketMQ

天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况。

RoketMQ在稳定性上可能更值得信赖，这些业务场景在阿里双11已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择RocketMQ。

3.RabbitMQ

RabbitMQ :结合erlang语言本身的并发优势，性能较好，社区活跃度也比较高，但是不利于做二次开发和维护。不过，RabbitMQ的社区十分活跃，可以解决开发过程中遇到的bug。

如果你的数据量没有那么大，小公司优先选择功能比较完备的RabbitMQ。

Kafka学习之路 （一）Kafka的简介

目录 一、简介 1.1　概述 1.2　消息系统介绍 1.3　点对点消息传递模式 1.4　发布-订阅消息传递模式 二、Kafka的优点 2.1　解耦 2.2　冗余（副本） 2.3　扩展性 2.4　灵活性&峰值处理能力 2.5　可恢复性 2.6　顺序保证 2.7　缓冲 2.8　异步通信 三、常用Message Queue对比 3.1　RabbitMQ 3.2　Redis 3.3　ZeroMQ 3.4　ActiveMQ 3.5　Kafka/Jafka 四、Kafka中的术语解释 4.1　概述 4.2　broker 4.3　Topic 4.3　Partition 4.4　Producer 4.5　Consumer 4.6　Consumer Group 4.7　Leader 4.8　Follower   正文 回到顶部 一、简介 1.1　概述 Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。 主要应用场景是：日志收集系统和消息系统。 Kafka主要设计目标如下： 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。 同时支持离线数据处理和实时数据处理。 Scale out:支持在线水平扩展 1.2　消息系统介绍 一个消息系统负责将数据从一个应用传递到另外一个应用，应用只需关注于数据，无需关注数据在两个或多个应用间是如何传递的。分布式消息传递基于可靠的消息队列，在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。 1.3　点对点消息传递模式 在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序。这种架构描述示意图如下： 生产者发送一条消息到queue，只有一个消费者能收到。 1.4　发布-订阅消息传递模式 在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。该模式的示例图如下： 发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。 回到顶部 二、Kafka的优点 2.1　解耦 在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。 2.2　冗余（副本） 有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。 2.3　扩展性 因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。 2.4　灵活性&峰值处理能力 在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。 2.5　可恢复性 系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。 2.6　顺序保证 在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。 2.7　缓冲 在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。 2.8　异步通信 很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。 回到顶部 三、常用Message Queue对比 3.1　RabbitMQ RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。 3.2　Redis Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。 3.3　ZeroMQ ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。 3.4　ActiveMQ ActiveMQ是Apache下的一个子项目。 […]

Kafka

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

rabbitmq

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

RabbitMQ消息持久化

目录 一、前言 二、队列持久化 　　2.1 查看存在的队列和消息数量 　　2.2 持久化队列 三、消息持久化 四、总结 一、前言 如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。当然还是会有一些小概率事件会导致消息丢失。 二、队列持久化 　　2.1 查看存在的队列和消息数量 在windows环境下，在rabbitmq的安装目录/sbin下，通过rabbitmqctl.bat list_queues查看　 这边启动了两个producer,分别生成两个队列hello 和 hello1,并且他们都有一个消息存在 重启rabbitmq，模拟故障 可以看到重启后两个队列都消失了 　　2.2 持久化队列 我们就hello队列持久化 在声明队列名称时，持久化队列，生产端和消费端都要 1 channel.queue_declare(queue=’hello', durable=True) 我们重复上面的操作，但是给hello队列做持久化，而hello1不做，并重启rabbitmq 可以看到重启后，hello队列还在，hello1队列消失了，但是原本hello中的一条消息也没有保存下来。所以在这边我们仅仅做到了消息队列的持久化，还没有做消息持久化。 三、消息持久化 我们刚才实现了在rabbitmq崩溃的情况下，就队列本身保存下来，重启后队列还在。接下来我们要将消息也保存下来，即消息的持久化 1 2 3 4 5 6 7 8 9 channel.basic_publish(exchange=",                       routing_key=’hello',                       body=’hello',                       properties=pika.BasicProperties(                           delivery_mode=2,  # make message persistent                       )) # 增加properties，这个properties 就是消费端 callback函数中的properties # delivery_mode = 2  持久化消息 生产端生成一个消息，并重启rabbitmq 可以看到，经过队列和消息持久化后的hello, 在重启的情况下，队列和消息都存在，没有消失。 消费端再重启后也能正常接收 四、总结 队列持久化需要在声明队列时添加参数 durable=True，这样在rabbitmq崩溃时也能保存队列 仅仅使用durable=True ，只能持久化队列，不能持久化消息 消息持久化需要在消息生成时，添加参数 properties=pika.BasicProperties(delivery_mode=2)   from: https://www.cnblogs.com/bigberg/p/8195622.html#_label3

在.Net中跑RocketMQ_入门篇（下）

上一篇讲了如何再控制台将RocketMQ跑起来，本篇讲解，在asp.net mvc种跑起来，含（发布、订阅）。 因篇幅过长，本次将不挨个贴源码，直接展示目录，根据上一篇文章，进行相应的调整即可。 1、新建一个类库，将MQ公共部分提出来（源码在入门篇1中有讲解和截图）： 如： 此时需要注意的有两点： (1)在RocketMQClientManager中，尽量将IRocketMQManager做成单例，避免每发一次消息，就实例化一个对象造成资源浪费。 (2)经测试，在RocketMQClientManager中发送消息时，遇到并发可能会存在消息丢失问题，防止消息丢失，可以在发布的时候，进行加锁处理。 2、在asp.net mvc 项目中配置MQ信息（需要在App_Data中修改RocketMQ的相应配置，入门篇1中有讲解）。 3、在asp.net mvc Global种添加初始化RocketMQ和订阅代码（按Tag进行订阅，也可订阅多有的Tag）： 注意：FarseerBootstrapper.Create().Initialize(); //必须注册MQ服务，否则运行会抛出，未注册MQ服务啥。 4、接下来可以在控制器调用RocketMQ,进行测试（发布消息的Tag为BootsShuTestTag,消费可以根据指定的Tag进行消费）： 发布、订阅运行结果： 注意、注意、注意： 1.需要将项目统一编译成X64、并且将项目寄宿到IIS,里边即可正常运行。在VS里边运行，是跑不起来的！ 2.作者不推荐在服务端进行订阅，如果在IIS上运行，则可以将订阅模块提出来写成windows服务。 大家根据自己的业务场景自行甄别及其使用。 以上如有不对的，可以拍砖，相互学习，共同进步。 欢迎加入.NET CORE/ASP.NET CORE 技术交流群，我们期待你的加入。 群号：702566187   from:https://blog.csdn.net/weixin_37207795/article/details/81023914

在.Net中跑RocketMQ_入门篇（上）

下面我将为大家介绍一款支持.NET端的RocketMQ。大名为：Farseer.Net.MQ.RocketMQ。 支持.Net Framework 4.5和.NetStandard 2.0 1、先创建一个.Net Framework的控制台，热热手。 2、通过Nuget安装相应组件。 Install-Package Farseer.Net.MQ.RocketMQ -Version 2.0.1 Install-Package Farseer.Net.Configuration 3、将ONSClient4CPP放到项目根目录： 如： 3.1、 3.2、 4、新建启动模块：PushModule： 如： 5、在我们创建的控制台中，编写RocketMQ代码： 如： 6、将项目改为X64。 7、修改App_Data下的Farseer.Net.josn(注意：程序运行，会自动生成改json配置文件) 8、运行控制台，能跑起来了吧！ 欢迎加入.NET CORE/ASP.NET CORE 技术交流群，我们期待你的加入。 群号：702566187   from:https://blog.csdn.net/weixin_37207795/article/details/81023757