kafka消息队列（Kafka消息队列反诈系统应用）

本篇文章给大家谈谈kafka消息队列，以及Kafka消息队列反诈系统应用对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

• 1、消息队列（kafka/nsq 等）与任务队列（celery）到底有什么不同
• 2、Kafka 设计详解之队列
• 3、soulcoder——消息队列知识总结(偏向于 Kafka)
• 4、消息队列之zeroMQ、rabbitMQ、kafka
• 5、Kafka，Mq和Redis作为消息队列使用
• 6、消息队列(三)kafka的一致性和失败处理策略

消息队列（kafka/nsq 等）与任务队列（celery）到底有什么不同

在和同事介绍celery时 同事说了一句“这不就是kafka吗？”。

那么YTask和nsq，celery和kafka？他们之间到底有什么不同呢？下面我结合自己的派掘理解。简单的分析一下郑凳，如有不足请指出。

首先喊羡旅，nsq和kafka它们属于消息队列 celery它们属于任务队列。

其实他们都是利用队列（先进先出）的特性已经增加自己的存储功能 实现不同场景下的处理

消息队列和任务队列，最大的不同之处就在于理念的不同 -- 消息队列传递的是“消息”，任务队列传递的是“任务” 。

这句话何解呢？

我们可以放到具体的应用场景上:

从上面的例子可看出:

因此可以简单认为任务队列就是消息队列在异步任务场景下的深度化定制开发。

Kafka 设计详解之队列

在 上文 中我们介绍了 Kafka 的网络通信，本文打算详细分析 Kafka 的核心 — 队列 的设计和实现，来对 Kafka 进行更深一步的了解。

队列是一种先进先出（FIFO）的数据结构，它是 Kafka 中最重要的部分，负责收集生产者生产的消息，并将这些消息传递给消费者。要实现一个队列有多种方式，Kafka 作为一个消息队列中间件，在设计队列时主要要考虑两个问题：

乍一看到这个问题，我们会想，内存的读取速度远快于磁盘，如果追求性能，内存也充足的话，当然是将生产者产生的消息数据写到内存（比如用一个数组或者链表来存储队列数据），供消费者消费。真的是这样吗？

下面我们依次分析下写内存和写磁盘文件的优缺链锋点，首先，内存的优点是读写速度非常快，但是，如果我们的目标是设计「大数据量」下的「高吞吐量」的消息队列，会有以下几个问题：

接下来我们来分析一下磁盘，写磁盘文件方式存储队列数据的优点就是能规避上述内存的缺点，但其有很严重的缺点，就是读写速度慢，如果纯依靠磁盘，那消息队列肯定做不到「高吞吐量」这个目标。

分析了内存跟磁盘的优缺点，好像我们还是只能棚告晌选写内存，但我们忽视了磁盘的两个情况：一是磁盘慢是慢在随机读写，如果是顺序读写，他的速度能达到 600MB/sec（RAID-5 磁盘阵列），并不慢，如果我们尽可能地将数据的读写设计成顺序的，可以大大提升性能。二是 现代的操作系统会（尽可能地）将磁盘里的文件进行缓存 。

有了操作系统级别的文件缓存，那用磁盘存储队列数据的方式就变得有优势了。首先，磁盘文件的数据会有文件缓存，所以不必担心随机读写的性能；其次，同样是使用内存，磁盘文件使用的是操作系统级别的内存，相比于在 Java 内存堆中存储队列，它没有 GC 问题，也没有 Java 对象的额外内存开销，更可以规避应用重启后的内存 load 数据耗时的问题，而且，文件缓存是操作系统提供的，因为我们只要简单的写磁盘文件，系统复杂性大大降低。

因此，Kafka 直接使用磁盘来存储消息队列的数据。

刚才我们已经决定用磁盘文件来存储队列数据，那么要如何选择数据结构呢？一般情况下，如果需要查找数据并随机访问，我们会用 B+ 树来存储数据，但其时间复杂度是 O(log N)，由于我们设计的是消息队列，我们可以完全顺序的写收到的生产者消息，消费者消费时，只要记录下消费者当前消费的位置，往后消费就可以了，这样可以对文件尽可能的进行顺序读写，同时，时间复杂度是O(1)。其实，这跟我们写日志的方式很像，每条日志顺序 append 到日志文件。

之前我们已经确定采用直接顺序写磁盘文件的方式来存储队列数据，下面我们来剖析下具体的实现细节。

在 Kafka 中，用一个文件夹存储一条消息队列，成为一个 Log，每条消息队列由多个文件组成，每个文件称为一个 LogSegment，每当一个 LogSegment 的大小到达阈值，系统就会重新生成一个 LogSegment；当旧的 LogSegment 过期需要清理时（虽然磁盘空间相对于内存会宽裕很多，我们可以保存更长时间的消息数据，比如一周，以供消费者更灵活的使用，但还是需要定期清理太老的数据），系统会根据清理策略删除这些文件。

现在我们知道一个队列（Log）是由多个队列段文件（LogSegment）组成的，那么 Kafka 是如何将这些文件逻辑上连接从而组成一条有序队列的呢？在生成每个队列段文件时，Kafka 用该段的初始位移来对其命名，如在新建一个队列时，友激会初始化第一个队列段文件，那么其文件名就是0，假设每个段的大小是固定值 L，那么第二个段文件名就是 L，第 N 个就是 （N - 1）* L。这样，我们就可以根据文件名对段文件进行排序，排序后的顺序就是整个队列的逻辑顺序。

了解了队列的基本实现，下面我们就来分析下队列的核心操作—读和写。

写操作发生在生产者向队列生产消息时，在上篇文章讲网络通信时我们已经说到，所有的客户端请求会根据协议转到一个 Handler 来具体处理，负责写操作的 Handler 叫 ProducerHandler，整个写请求的流程如下：

之前我们说过，如果是顺序写，由于省掉了磁头寻址的时间，磁盘的性能还是很高的，我们看到 Kakfa 队列是以顺序方式写的，所以性能很高。但是，如果一台 Kafka 服务器有很多个队列，而硬盘的磁头是有限的，所以还是得在不同的队列直接来回切换寻址，性能会有所下降。

队列的读操作发送在消费者消费队列数据时，由于队列是线性的，只需要记录消费者上次消费到了哪里（offset），接下去消费就好了。那么首先会有一个问题，由谁来记消费者到底消费到哪里了？

一般情况下，我们会想到让服务端来记录各个消费者当前的消费位置，当消费者来拉数据，根据记录的消费位置和队列的当前位置，要么返回新的待消费数据，要么返回空。让服务端记录消费位置，当遇到网络异常时会有一些问题，比如服务端将消息发给消费者后，如果网络异常消费者没有收到消息，那么这条消息就被「跳过」了，当然我们可以借鉴二阶段提交的思想，服务端将消息发送给消费者后，标记状态为「已发送」，等消费者消费成功后，返回一个 ack 给服务端，服务端再将其标记为「成功消费」。不过这样设计还是会有一个问题，如果消费者没有返回 ack 给服务端，此时这条消息可能在已经被消费也可能还没被消费，服务端无从得知，只能根据人为策略跳过（可能会漏消息）或者重发（可能存在重复数据）。另一个问题是，如果有很多消费者，服务端需要记录每条消息的每个消费者的消费状态，这在大数据的场景下，非常消耗性能和内存。

Kafka 将每个消费者的消费状态记录在消费者本身（隔一段时间将最新消费状态同步到 zookeeper），每次消费者要拉数据，就给服务端传递一个 offset，告诉服务端从队列的哪个位置开始给我数据，以及一个参数 length，告诉服务端最多给我多大的数据（批量顺序读数据，更高性能），这样就能使服务端的设计复杂度大大降低。当然这解决不了一致性的问题，不过消费者可以根据自己程序特点，更灵活地处理事务。

下面就来分析整个读的流程：

分布式系统中不可避免的会遇到一致性问题，主要是两块：生产者与队列服务端之间的一致性问题、消费者与队列服务端之间的一致性问题，下面依次展开。

当生产者向服务端投递消息时，可能会由于网络或者其他问题失败，如果要保证一致性，需要生产者在失败后重试，不过重试又会导致消息重复的问题，一个解决方案是每个消息给一个唯一的 id，通过服务端的主动去重来避免重复消息的问题，不过这一机制目前 Kafka 还未实现。目前 Kafka 提供配置，供用户不同场景下选择允许漏消息（失败后不重试）还是允许重复消息（失败后重试）。

由于在消费者里我们可以自己控制消费位置，就可以更灵活的进行个性化设计。如果我们在拉取到消息后，先增加 offset，然后再进行消息的后续处理，如果在消息还未处理完消费者就挂掉，就会存在消息遗漏的问题；如果我们在拉取到消息后，先进行消息处理，处理成功后再增加 offset，那么如果消息处理一半消费者挂掉，会存在重复消息的问题。要做到完全一致，最好的办法是将 offset 的存储与消费者放一起，每消费一条数据就将 offset+1。

本文介绍了 Kafka 的队列实现以及其读写过程。Kafka 认为操作系统级别的文件缓存比 Java 的堆内存更省空间和高效，如果生产者消费者之间比较「和谐」的话，大部分的读写操作都会落在文件缓存，且在顺序读写的情况下，硬盘的速度并不慢，因此选择直接写磁盘文件的方式存储队列。在队列的读写过程中，Kafka 尽可能地使用顺序读写，并使用零拷贝来优化性能。最后，Kafka 让消费者自己控制消费位置，提供了更加灵活的数据消费方式。

[img]

soulcoder——消息队列知识总结(偏向于 Kafka)

[toc]

分析一个消息队列主要从这几个点出来。

在后半部分主要分析了 kafka 对以上几点的保证。

详见下文分析重点分析。

事务支持方面，ONS/RocketMQ较为优秀，但是不支持消息批量操作, 不保证消息至少被消费一次.

Kafka提供完全分布式架构, 并有replica机制, 拥有较高的可用性和可靠性, 理论上支持消息无限堆积, 支持批量操作, 消费者采用Pull方式获取消息, 消息有序, 通过控制能够保证所有消息被消费且仅被消费一次. 但是官方提供的运维工具不友好，开源社区的运维工具支持的版本一般落后于最新版本的Kafka.

目前使用的MNS服务，拥有HTTP REST API, 使用简单, 数据可靠性高, 但是不保证消息有序，不能回溯数据.

RabbitMQ为重量级消息系统， 支持多协议(很多协议是目前业务用不到的), 但是不支持回溯数据, master挂掉之后， 需要手动从slave恢复, 可用性略逊一筹.

以rcoketMQ为例，他的集群就有

第一眼看到这个图，就觉得和kafka好像，只是NameServer集群，在kafka中是用zookeeper代替，都是用来保存和发现master和slave用的。

通信过程如下:

Producer 与 NameServer集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Broker Master 建立长连接，且定时向 Broker 发送心跳。

Producer 只能将消息发送到 Broker master，但是 Consumer 则不一样，它同时和提供 Topic 服务的 Master 和 Slave建立长连接，既可以从 Broker Master 订阅消息，也可以从 Broker Slave 订阅消息。

那么kafka呢？

为了对比说明直接上kafka的拓补架构图

如上图所示，一个典型的Kafka集群中包含若干Producer（可以扰族迹是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息缓并发布到broker，Consumer使用pull模式从broker订阅并消费消息。

最骚的一个操作，消费者业务自己去保证幂等性。

换一个说法，如何保证消息队列的幂等性?

另外说一点，幂等性的保证需要在一次请求中所有链路都是幂等的，再能最终保证这次请求的幂等，比如前段按钮点击两次，后端认为都是这是两次不同的请求，当然处理成两次请求，所以说一个请求的幂等性，需要全局的幂等才能保证。

其实无论是哪种消息队列，造成重复消费原因其实都是类似的。正常情况下，消费者在消费消息时候，消费完毕后，会发送一个确认信息给消息队列，消息队列就知道该消息被消费了，就会将该消息从消息队列中删除。只是不同的消息队列发送的确认信息形式不同。

例如RabbitMQ是发送一个ACK确认消息，RocketMQ是返回一个CONSUME_SUCCESS成功标志，kafka实际上有个offset的概念，简单说一下(后续详细解释),就是每一个消息都有一个offset，kafka消费过消息后，需要提交offset，让消息队列知道自己穗圆已经消费过了。

那造成重复消费的原因?，就是因为网络传输等等故障，确认信息没有传送到消息队列，导致消息队列不知道自己已经消费过该消息了，再次将该消息分发给其他的消费者。

如何解决?这个问题针对业务场景来答分以下几点

其实这个可靠性传输，每种MQ都要从三个角度来分析:生产者弄丢数据、消息队列弄丢数据、消费者弄丢数据。

从生产者弄丢数据这个角度来看，RabbitMQ提供transaction和confirm模式来确保生产者不丢消息。

transaction(事物机制)机制就是说，发送消息前，开启事物(channel.txSelect())，然后发送消息，如果发送过程中出现什么异常，事物就会回滚(channel.txRollback())，如果发送成功则提交事物(channel.txCommit())。然而缺点就是吞吐量下降了。

生产上用confirm模式的居多。一旦channel进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，rabbitMQ就会发送一个Ack给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了.如果rabiitMQ没能处理该消息，则会发送一个Nack消息给你，你可以进行重试操作。

简单来讲 confirm模式就是生产者发送请求，到了消息队列，消息队列会回复一个消息收到的应答，如果没收到，生产者开始重试。

处理消息队列丢数据的情况，一般是开启持久化磁盘的配置。这个持久化配置可以和confirm机制配合使用，你可以在消息持久化磁盘后，再给生产者发送一个Ack信号。这样，如果消息持久化磁盘之前，rabbitMQ阵亡了，那么生产者收不到Ack信号，生产者会自动重发。

消费者丢数据一般是因为采用了自动确认消息模式。这种模式下，消费者会自动确认收到信息。这时rahbitMQ会立即将消息删除，这种情况下如果消费者出现异常而没能处理该消息(但是消息队列那边已经认为消息被消费了)，就会丢失该消息。

至于解决方案，采用手动确认消息即可。

kafka为例

Producer在发布消息到某个Partition时，先通过ZooKeeper找到该Partition的Leader，然后无论该Topic的Replication Factor为多少（也即该Partition有多少个Replica），Producer只将该消息发送到该Partition的Leader。Leader会将该消息写入其本地Log。每个Follower都从Leader中pull数据。

在kafka生产中，基本都有一个leader和多个follwer。follwer会去同步leader的信息。因此，为了避免生产者丢数据，做如下两点配置

针对消息队列丢数据的情况，无外乎就是，数据还没同步，leader就挂了，这时zookpeer会将其他的follwer切换为leader,那数据就丢失了。针对这种情况，应该做两个配置。

这种情况一般是自动提交了offset，然后你处理程序过程中挂了。kafka以为你处理好了。再强调一次offset是干嘛的。

offset：指的是kafka的topic中的每个消费组消费的下标。简单的来说就是一条消息对应一个offset下标，每次消费数据的时候如果提交offset，那么下次消费就会从提交的offset加一那里开始消费。

比如一个topic中有100条数据，我消费了50条并且提交了，那么此时的kafka服务端记录提交的offset就是49(offset从0开始)，那么下次消费的时候offset就从50开始消费。

针对这个问题，通过某种算法，将需要保持先后顺序的消息放到同一个消息队列中(kafka中就是partition,rabbitMq中就是queue)。然后只用一个消费者去消费该队列。

有的人会问:那如果为了吞吐量，有多个消费者去消费怎么办？

简单来说消息的时序性也可以通过错误重试来解决。

比如我们有一个微博的操作，发微博、写评论、删除微博，这三个异步操作。如果是这样一个业务场景，那只要重试就行。比如你一个消费者先执行了写评论的操作，但是这时候，微博都还没发，写评论一定是失败的，等一段时间。等另一个消费者，先执行写评论的操作后，再执行，就可以成功。

总之，针对这个问题，我的观点是保证入队有序就行，出队以后的顺序交给消费者自己去保证，没有固定套路。

为了做到水平扩展，一个topic实际是由多个partition组成的，遇到瓶颈时，可以通过增加partition的数量来进行横向扩容。

单个parition内是保证消息有序。

订阅topic是以一个消费组来订阅的，一个消费组里面可以有多个消费者。

同一个消费组中的两个消费者，只能消费一个partition。

换句话来说，就是一个partition，只能被消费组里的一个消费者消费，但是可以同时被多个消费组消费。

如果消费组内的消费者如果比partition多的话，那么就会有个别消费者一直空闲。

kafka api 提供了很多功能比如

生产者能指定 topic 和 Partition 来投递消息，并且还有延迟消息，事务消息等等，详见下面的 api 文档

这个是 api 的中文文档

Kakfa Broker集群受Zookeeper管理。

这里先说下

关于partition的分配，还有leader的选举，总得有个执行者。在kafka中，这个执行者就叫controller。kafka使用zk在broker中选出一个controller，用于partition分配和leader选举。

所有的Kafka Broker节点一起去Zookeeper上注册一个临时节点，并且只有一个Kafka Broker会注册成功，其他的都会失败，所以这个成功在Zookeeper上注册临时节点的这个Kafka Broker会成为 Kafka Broker Controller ，其他的Kafka broker叫 Kafka Broker follower 。（这个过程叫Controller在ZooKeeper注册Watch）。

这个Controller会监听其他的Kafka Broker的所有信息，如果这个kafka broker controller宕机了，在zookeeper上面的那个临时节点就会消失，此时所有的kafka broker又会一起去Zookeeper上注册一个临时节点。

Kafka提供3种消息传输一致性语义：最多1次，最少1次，恰好1次。

最少1次(at most once)：可能会重传数据，有可能出现数据被重复处理的情况;

最多1次(at least once)：可能会出现数据丢失情况;

恰好1次(Exactly once)：并不是指真正只传输1次，只不过有一个机制。确保不会出现“数据被重复处理”和“数据丢失”的情况。

操作系统本身有一层缓存，叫做page cache，是在内存里的缓存，我们也可以称之为os cache，意思就是操作系统自己管理的缓存。

每新写一条消息，kafka就是在对应的文件append写，所以性能非常高。

我觉得的靠的是这两个参数

这篇主要从生产和消费的角度详细给出的过程

消息队列之zeroMQ、rabbitMQ、kafka

      首先消息是网络通讯的载体，队列可以理解是一种先进先出的数据结构，消息队列是存放消息的容器，是分布式系统中的重要组件。消息队列的优势在于：解耦、异步、削峰，把相关性不

强的模块独立分开视为解耦，异步就是非必要逻辑异步方式处理，加快响应速度，削峰是避免短期高并发导致系统问题进行缓冲队列处理。消息队列的缺点在于：加强系统复杂性、系统可用性降低，使

用了消息队列系统出现问题排查的范围就变大、需要考虑消息队列导致的问题。

          本文说明主流的消息队列，针对使用过的zeroMQ和rabbitMQ、Kakfa：

          zeroMQ ：C语言开发，号称最快的消息队列，本着命名zero的含义，中油中间架构使用简单，表面上是基于socket的封装套接字API，在多个节点应用场景下非常灵活、架构的可扩展性很强，

实现N到M的协同处理；

            zmq的socket模式： req、rep、push、pull、pub、sub、router、dealer。

       竖纳   （1）req和rep：请求回应模型，req和rep都可以请求誉举和回答，不同的只是req是先send再rec，rep是先rec再send。支持N个请求端一个接收端，也支持N个接收端一个请求端。N个接收端采

用rr负载均衡。 哪个是“一”端，哪个就bind端口，“N”端就只能connect，所以，req+rep无论谁bind端口，肯定要有一个是“一”。

          （2）  router和dealer：随时可以发送和接收的req和rep，看起来router+dealer跟 req+rep属于同类功能。因为router和dealer可以随时发送接收，所以它们可以用来做路由。一个router用来响

应N个req，然后它在响应处理的时候，再通过另一个socket把请求扔出去，接收者是另外的M个rep，这就做到N:M。

         （3）pub和sub ：订阅和推送，对应发布者和订阅者。

         （4）push和pull：就是管道，一个只推数据，一个只拉数据。

           rabbitMQ ：使用erlang语言开发，高并发特点，基于AMQP（即Advanced Message Queuing Protocol）的开源高级消费队列，AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/

订阅）、可靠性、安全），企业级适应性和稳定性，并且有WEB管理界面方便用户查看和管理。以下是rabbitMQ的结构图：

         （1）Producer:数据发送方，一般一个Message有两个部分：payload（有效载荷）和label（标签），payload是数据实际载体，label是exchange的名字或者一个tag，决定发给哪个Consumer；

         （2）Exchange: 内部 消息交换器,exchange从生产者那收到消息后，一般会指定一个Routing Key，来指定这个消息的路由规则，当然Routing Key需要与Exchange Type及Binding key联合使用

才能最终生效，根据路由规则，匹配查询表中的routing key，分发消息到queue中；

         （3）binding：即绑定，绑定（Binding）Exchange与Queue的同时，一般会指定一个Binding key，但不一定会庆纤碧生效，依赖于Exchange Type；

         （4）Queue：即队列是rabbitmq内部对象，用于存储消息，一个message可以被同时拷贝到多个queue中，queue对load balance的处理是完美的。对于多个Consumer来说，RabbitMQ 使用循

环的方式（round-robin）的方式均衡的发送给不同的Consumer；

         （5）Connection与Channel： Connection 就是一个TCP的连接，Producer和Consumer都是通过TCP连接到RabbitMQ Server， Channel 是为了节省开销建立在上述的TCP连接中的接口，大部

分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等；

        （6）Consumer：即数据的接收方，如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者；

        （7）Broker: 即RabbitMQ Server，其作用是维护一条从Producer到Consumer的路线，保证数据能够按照指定的方式进行传输；

       （8）Virtual host：即虚拟主机，当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue；

         rabbitMQ消息转发中的路由转发是重点，生产者Producer在发送消息时，都需要指定一个RoutingKey和Exchange，Exchange收到消息后可以看到消息中指定的RoutingKey，再根据当前

Exchange的ExchangeType,按一定的规则将消息转发到相应的queue中去。三种Exchage type：

       （1）Direct exchange ：直接转发路由，原理是通过消息中的routing key，与binding 中的binding-key 进行比对，若二者匹配，则将消息发送到这个消息队列；

          比如：消息生成者生成一个message(payload是1，routing key为苹果)，两个binding(binding key分别为苹果、香蕉);exchange比对消息的routing key和binding key后，将消息发给了queue1，消息消费者1获得queue1的消息；

       （2）Topic exchange： 通配路由，是direct exchange的通配符模式，

          比如：消息生成者生成一个message(payload是1，routing key为quick.orange.rabbit)，两个binding(binding key分别为*.orange. 、 *.*.rabbit)；exchange比对消息的routing key和binding key

后,exchange将消息分发给两个queue，两个消费者获得queue的消息；

     （3）Fanout exchange： 复制分发路由，原理是不需要routkey，当exchange收到消息后，将消息复制多份转发给与自己绑定的消息队列，

          比如：消息生成者生成一个message(payload是1，routing key为苹果)，两个binding(binding key分别为苹果、香蕉)；exchange将消息分发给两个queue，两个消费者获得queue的消息；

       rabbiMQ如何保证消息的可靠性？

     （1）Message durability：消息持久化，非持久化消息保存在内存中，持久化消息写入内存同时也写入磁盘；

     （2）Message acknowledgment：消息确认机制，可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移

除。通过ACK。每个Message都要被acknowledged（确认，ACK）。

     （3）生产者消息确认机制：AMQP事务机制、生产者消息确认机制（publisher confirm）。

     最后， 对比一下zeroMQ、rabbitMQ、kafka主流的消息队列的性能情况：

      对比方向                                                                          概要

      吞吐量                             万级 RabbitMQ 的吞吐量要比 十万级甚至是百万级Kafka 低一个数量级。ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。

      可用性                            都可以实现高可用。RabbitMQ 都是基于主从架构实现高可用性。 kafka 也是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用

      时效性                             RabbitMQ 基于erlang开发，所以并发能力很强，性能极其好，延时很低，达到微秒级。其他两个个都是 ms 级。

      功能支持                          Kafka 功能较为简单，主要支持简单的MQ功能，在大数据领域实时计算以及日志采集被大规模使用；ZeroMQ能够  实现RabbitMQ不擅长的高级/复杂 的队列

      消息丢失                          RabbitMQ有ack模型，也有事务模型，保证至少不会丢数据，  Kafka 理论上不会丢失，但不排除批量情况下。

      开发环境                          RabbitMQ需要erlang支持、kafka基于zookeeper管理部署、zeroMQ程序编译调用即可

      封装库                               基于c++开发，使用RabbitMQ-C，cppKafka，而zeroMQ基于C语言开发，无需封装

Kafka，Mq和Redis作为消息队列使用

kafka是个日志处理缓冲组件，在大数据信息处理中使用。和传统的消息队列相比较简化了队列结构和功能，以流形式处理存储（持久化）消息（主要是日志）。日志数据量巨大，处理组件一般会处理不过来，所以作为孝衫神缓冲层的kafka，支持巨大吞吐量。为了防止信息丢失，其消息被调用后不直接丢弃，要多存储一段时间，等过期时间过了才丢弃。这是mq和redis不能具备的。主要特点如下:巨型存储量: 支持TB甚至PB级别数据。高吞吐，高IO：一般配置的服务器能实现单机每秒100K以上消息的传输。消息分区，分布式消费：能保消息顺序传输。 支持离线数据处理和实时数据处理。Scale out：支持在线水平扩展，以支持更大数塌洞据处理量

redis只是提供一个高性能的、原子操作内存键值对，具有高速访问能力，可用做消息队列的存储，但是不具备消息队列的任何功能和逻辑，要作为消息队列来实现的话，功能和逻辑要通过上层应用自己实现。

我们以RabbitMQ为例介绍。它是用Erlang语言开发的开源的消息队列，支持多种协议，包括AMQP，XMPP, SMTP, STOMP。适合于企业级的开发。

MQ支持Broker构架，消息发送给客户端时需要在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。

还有ActiveMq，ZeroMq等。功能基本上大同小异。并发吞吐TPS比较，ZeroMq 最好，RabbitMq 次之， ActiveMq 最差巧亏。

原文：

消息队列(三)kafka的一致性和失败处理策略

    如前篇所讲，假如对 登录系统，主流程是：用户请求--验证账户密码--返回成功；一些其他流程包括：积分、提醒、荣誉系统等；这里有个问题是：如果这两者有一个失败了怎么办？

    由于Message由Producer发送到Topic的Master Partition的时候，为了保证可靠性，会等待In-Sync的Replicas都同步完成之后才会返回成功；这里同样有个问题：如果返回超时怎么办？

   在Consumer的客户端，为了提高效率，会分为Fetch线程和Consume线程，在consume完成判链备之后，会提交offset；问题是：提交offset和consume仅有一个失败怎么办？

    这种语义，很难被接受，网络故障、主副本选举、超时等因素就可能造成这种现象，而且若没这方面的意识，日志可能都不会打印，人工补偿都没法做到，造成这种现象的掘毁原因有：

a) 生产者：(很多参数可以影响，挑几个重点的)

      异步调用，并没有设置回调函数；

      ack=0,不等Broker确认就继续发送消息;

      ack=1,等待leader确认后再发送消息,若follower没有跟上, 且leader挂掉,再选举将丢失

      retries=0，不重试，或者重试之后，仍旧失败，不考虑重试队列或者人工补偿

b) 消费者

    设置为自动提交，时间间隔设置的较短，且不手动提交offset；前文说到，Consumer分为Fetch和Consume两步，自动提交的offset是Fetch的，所以提交的最高offset的message还未处理；

    这种语义，被实现的较多，往往Consumer会拉到重复的message，再去在cache中做一层去重的处理，然后实现Exactly Once，至少在0.11之前未实现kafka事务之前；

a) 生产者 

   ack=ALL/-1,即等待In-sync的follower确认，才继续发送；

  且retries 1，且还是失败唤毁了之后，要有重试队列、死信队列、人工补偿的方案；

b)消费者  ：关闭自动提交，并每次手动提交offset

   a) 在At least once 的基础上，在业务方做幂等或者去重，比如redis去重，或业务上幂等;

   b) 在At least once 的基础上，将offset和处理的结果组成一个原子事务，但这种方法，我个人觉得引入了更多的复杂性，参考

   c) 上面保证了不出错的情况下的，出错的情况下，需要重试队列、死信队列进行补偿；

   d) kafka的0.11版本之后，提供了事务，利用事务可以实现；

      kafka中没有实现重试队列和死信队列的功能，但是由于当前的message的offset如果不提交，就会阻塞后续的消费，所以需要预留失败的message补偿的机制；实现方法有几种：

       用本地队列去控制，设置在定时器中，给任务设置30s，5min，30min三次重试的机会，如果不行，持久化到DB中，进行人工干预，当然报警、日志都要跟上; 优点：逻辑简单、实现简单，缺点：需要一些机制保证本地cache的可靠性，比如加hook预防服务更新，但是这样仍不能完全解决，还要面临初始化重试队列、宕机来不及调用hook等问题；

    这个代价略大，而且这种不太受我们控制，很多消息队列的选型是已经固定的；修改kafka实现也不可能；

    这种思路也面临很大的问题：就是kafka没有实现延时的功能，那么新的message可能瞬时就被消费了，但是这个时候导致失败的原因，比如DB连接、网络问题没有解决，很快尝试几次之后，就进到死信队列了；

    还有一种是《深入kafka》一书中也讲了一些方法，但是都需要改动kafka的内部实现，这个不太适合小型项目；

   redis是很普遍的组件，而且对于少量失败、且对重试时间要求没那么严格的情况下，redis很适合，redis的持久化、哨兵之类的，对可靠性保证的还是很不错的；

留一篇文章，留着回忆redis的时候，再写吧

关于kafka消息队列和Kafka消息队列反诈系统应用的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。