标签：数据同步 rom 重试 leo ble 流处理 文件系统 行数据 发送
1. kafka概述Apache Kafka 是一个快速、可扩展的、高吞吐的、可容错的分布式“发布-订阅”消息系统， 使用 Scala 与 Java 语言编写，能够将消息从一个端点传递到另一个端点，较之传统的消息中 间件(例如 ActiveMQ、RabbitMQ)，Kafka 具有高吞吐量、内置分区、支持消息副本和高容 错的特性，非常适合大规模消息处理应用程序。
Kafka 官网: http://kafka.apache.org/
Kafka主要设计目标如下：
Kafka通常用于两大类应用程序：
要了解Kafka如何执行这些操作，让我们从头开始深入研究Kafka的功能。
首先几个概念：
kafka的应用场景非常多, 下面我们就来举几个我们最常见的场景
用户在网站的不同活动消息发布到不同的主题中心，然后可以对这些消息进行实时监测、实时处理。当然，也可以加载到Hadoop或离线处理数据仓库，对用户进行画像。像淘宝、天猫、京东这些大型电商平台，用户的所有活动都要进行追踪的。
Kafka与其他MQ相比，最大的特点就是高吞吐率。为了增加存储能力，Kafka将所有的消息都写入到了低速大容量的硬盘。按理说，这将导致性能损失，但实际上，Kafka仍然可以保持超高的吞吐率，并且其性能并未受到影响。其主要采用如下方式实现了高吞吐率。
在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。
因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。
在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。
在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。
很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。
RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。
Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。
ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。
ActiveMQ是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。
Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。
任何允许发布与使用无关的消息发布的消息队列都有效地充当了运行中消息的存储系统。Kafka的不同之处在于它是一个非常好的存储系统。
写入Kafka的数据将写入磁盘并进行复制以实现容错功能。Kafka允许生产者等待确认，以便直到完全复制并确保即使写入服务器失败的情况下写入也不会完成。
Kafka的磁盘结构可以很好地扩展使用-无论服务器上有50 KB还是50 TB的持久数据，Kafka都将执行相同的操作。
由于认真对待存储并允许客户端控制其读取位置，因此您可以将Kafka视为一种专用于高性能，低延迟提交日志存储，复制和传播的专用分布式文件系统。
Kafka的流概念与传统的企业消息传递系统相比如何？
传统上，消息传递具有两种模型：排队和发布-订阅。在队列中，一组使用者可以从服务器中读取内容，并且每条记录都将转到其中一个。在发布-订阅记录中广播给所有消费者。这两个模型中的每一个都有优点和缺点。排队的优势在于，它允许您将数据处理划分到多个使用者实例上，从而扩展处理量。不幸的是，队列不是多用户的—一次进程读取了丢失的数据。发布-订阅允许您将数据广播到多个进程，但是由于每条消息都传递给每个订阅者，因此无法扩展处理。
Kfka的消费者群体概念概括了这两个概念。与队列一样，使用者组允许您将处理划分为一组进程（使用者组的成员）。与发布订阅一样，Kafka允许您将消息广播到多个消费者组。
Kafka模型的优点在于，每个主题都具有这些属性-可以扩展处理范围，并且是多订阅者-无需选择其中一个。
与传统的消息传递系统相比，Kafka还具有更强的订购保证。
传统队列将记录按顺序保留在服务器上，如果多个使用者从队列中消费，则服务器将按记录的存储顺序分发记录。但是，尽管服务器按顺序分发记录，但是这些记录是异步传递给使用者的，因此它们可能在不同的使用者上乱序到达。这实际上意味着在并行使用的情况下会丢失记录的顺序。消息传递系统通常通过“专有使用者”的概念来解决此问题，该概念仅允许一个进程从队列中使用，但是，这当然意味着在处理中没有并行性。
Kafka做得更好。通过在主题内具有并行性（即分区）的概念，Kafka能够在用户进程池中提供排序保证和负载均衡。这是通过将主题中的分区分配给消费者组中的消费者来实现的，以便每个分区都由组中的一个消费者完全消费。通过这样做，我们确保使用者是该分区的唯一读取器，并按顺序使用数据。由于存在许多分区，因此仍然可以平衡许多使用者实例上的负载。但是请注意，使用者组中的使用者实例不能超过分区。
仅读取，写入和存储数据流是不够的，目的是实现对流的实时处理。
在Kafka中，流处理器是指从输入主题中获取连续数据流，对该输入进行一些处理并生成连续数据流以输出主题的任何东西。
例如，零售应用程序可以接受销售和装运的输入流，并输出根据此数据计算出的重新订购和价格调整流。
可以直接使用生产者和消费者API进行简单处理。但是，对于更复杂的转换，Kafka提供了完全集成的Streams API。这允许构建执行非重要处理的应用程序，这些应用程序计算流的聚合或将流连接在一起。
该功能有助于解决此类应用程序所面临的难题：处理无序数据，在代码更改时重新处理输入，执行状态计算等。
流API建立在Kafka提供的核心原语之上：它使用生产者和使用者API进行输入，使用Kafka进行状态存储，并使用相同的组机制来实现流处理器实例之间的容错。
主题。在 Kafka 中，使用一个类别属性来划分消息的所属类，划分消息的这个类称为 topic。 topic 相当于消息的分类标签，是一个逻辑概念
物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处
##2.2 Partition
分区。topic 中的消息被分割为一个或多个 partition，其是一个物理概念，对应到系统上 就是一个或若干个目录。partition 内部的消息是有序的，但 partition 间的消息是无序的。
段。将 partition 进一步细分为了若干的 segment，每个 segment 文件的大小相等。
Kafka 集群包含一个或多个服务器，每个服务器节点称为一个 broker。
broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。
如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。
如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。
生产者, 即消息的发布者. 生产者将数据发布到他们选择的主题。生产者负责选择将哪个记录分配给主题中的哪个分区。即: 生产者生产的一条消息，会被写入到某一个 partition。
##2.6 Consumer
消费者。可以从 broker 中读取消息。
一个消费者可以消费多个 topic 的消息
一个消费者可以消费同一个 topic 中的多个 partition 中的消息
一个 partiton 允许多个 consumer 同时消费
consumer group 是 kafka 提供的可扩展且具有容错性的消费者机制。组内可以有多个消 费者，它们共享一个公共的 ID，即 group ID。组内的所有消费者协调在一起来消费订阅主题 的所有分区。
Kafka 保证同一个 consumer group 中只有一个 consumer 会消费某条消息，实际上，Kafka 保证的是稳定状态下每一个 consumer 实例只会消费某一个或多个特定的 partition，而某个 partition 的数据只会被某一个特定的 consumer 实例所消费。
下面我们用官网的一张图, 来标识consumer数量和partition数量的对应关系
由两台服务器组成的Kafka群集，其中包含四个带有两个使用者组的分区（P0-P3）。消费者组A有两个消费者实例，组B有四个。
其实对于这个消费组, 以前一直搞不明白, 我自己的总结是:
topic中的partitoin到group是发布订阅的通信方式，即一条topic的partition的消息会被所有的group消费，属于一对多模式；group到consumer是点对点通信方式，属于一对一模式。
举个例子: 不使用group的话，启动10个consumer消费一个topic，这10个consumer都能得到topic的所有数据，相当于这个topic中的任一条消息被消费10次。
使用group的话，连接时带上groupid，topic的消息会分发到10个consumer上，每条消息只被消费1次
分区副本。副本是一个分区的备份，是为了防止消息丢失而创建的分区的备份。
每个 partition 有多个副本，其中有且仅有一个作为 Leader，Leader 是当前负责消息读写 的 partition。即所有读写操作只能发生于 Leader 分区上。
所有Follower都需要从Leader同步消息，Follower与Leader始终保持消息同步。Leader 与 Follower 的关系是主备关系，而非主从关系。
ISR，In-Sync Replicas，是指副本同步列表。 ISR列表是由Leader负责维护。
偏移量。每条消息都有一个当前Partition下唯一的64字节的offset，它是相当于当前分区第一条消息的偏移量。
Kafka集群的多个broker中，有一个会被选举controller，负责管理整个集群中partition和replicas的状态。
只有 Broker Controller 会向 zookeeper 中注册 Watcher，其他 broker 及分区无需注册。即 zookeeper 仅需监听 Broker Controller 的状态变化即可。
HW，HighWatermark，高水位，表示 Consumer 可以消费到的最高 partition 偏移量。HW 保证了 Kafka 集群中消息的一致性。确切地说，是保证了 partition 的 Follower 与 Leader 间数 据的一致性。
LEO，Log End Offset，日志最后消息的偏移量。消息是被写入到 Kafka 的日志文件中的， 这是当前最后一个写入的消息在 Partition 中的偏移量。
我相信你看完上面的概念还是懵逼的, 好吧, 下面我们就用图来形象话的表示两者的关系吧
Zookeeper 负责维护和协调 broker，负责 Broker Controller 的选举。
在 kafka0.9 之前版本，offset 是由 zk 负责管理的。
总结:zk 负责 Controller 的选举，Controller 负责 leader 的选举。
Coordinator一般指的是运行在每个broker上的group Coordinator进程，用于管理Consumer Group中的各个成员，主要用于offset位移管理和Rebalance。一个Coordinator可以同时管理多个消费者组。
当消费者组中的数量发生变化，或者topic中的partition数量发生了变化时，partition的所有权会在消费者间转移，即partition会重新分配，这个过程称为再均衡Rebalance。
再均衡能够给消费者组及broker带来高性能、高可用性和伸缩，但在再均衡期间消费者是无法读取消息的，即整个broker集群有小一段时间是不可用的。因此要避免不必要的再均衡。
##2.18 offset commit
Consumer从broker中取一批消息写入buffer进行消费，在规定的时间内消费完消息后，会自动将其消费消息的offset提交给broker，以记录下哪些消息是消费过的。当然，若在时限内没有消费完毕，其是不会提交offset的。
? 消息发送者将消息发送给broker, 并形成最终的可供消费者消费的log, 是已给比较复杂的过程
? 在通过 API 方式发布消息时，生产者是以 Record 为消息进行发布的。Record 中包含 key 与 value，value 才是我们真正的消息本身，而 key 用于路由消息所要存放的 Partition。消息 要写入到哪个 Partition 并不是随机的，而是有路由策略的。
若指定了 partition，则直接写入到指定的 partition;
若未指定 partition 但指定了 key，则通过对 key 的 hash 值与 partition 数量取模，该取模
结果就是要选出的 partition 索引;
如果 partition leader 接收到了新的消息， ISR 中其它 Follower 正在同步过程中，还未同 步完毕时 leader 宕机。此时就需要选举出新的 leader。若没有 HW 截断机制，将会导致 partition 中 leader 与 follower 数据的不一致。
当原 Leader 宕机后又恢复时，将其 LEO 回退到其宕机时的 HW，然后再与新的 Leader进行数据同步，这样就可以保证老 Leader 与新 Leader 中数据一致了，这种机制称为 HW 截断机制。
生产者向 kafka 发送消息时，可以选择需要的可靠性级别。通过 request.required.acks参数的值进行设置。
异步发送。生产者向 kafka 发送消息而不需要 kafka 反馈成功 ack。该方式效率最高，但可靠性最低。其可能会存在消息丢失的情况。
同步发送。生产者发送消息给 kafka，broker 的 partition leader 在收到消息后马上发送 成功 ack(无需等等 ISR 中的 Follower 同步)，生产者收到后知道消息发送成功，然后会再发送消息。如果一直未收到 kafka 的 ack，则生产者会认为消息发送失败，会重发消息。
该方式对于 Producer 来说，若没有收到 ACK，一定可以确认消息发送失败了，然后可以 重发;但是，即使收到了 ACK，也不能保证消息一定就发送成功了。故，这种情况，也可能 会发生消息丢失的情况。
同步发送。生产者发送消息给 kafka，kafka 收到消息后要等到 ISR 列表中的所有副本都 同步消息完成后，才向生产者发送成功 ack。如果一直未收到 kafka 的 ack，则认为消息发送 失败，会自动重发消息。该方式会出现消息重复接收的情况。
? 生产者将消息发送到topitc中, 消费者即可对其进行消费, 其消费过程如下:
当leader宕机后，broker controller会从ISR中挑选一个follower成为新的leader。如果ISR中没有其他副本怎么办？可以通过unclean.leader.election.enable的值来设置leader选举范围。
必须等到ISR列表中所有的副本都活过来才进行新的选举。该策略可靠性有保证，但可用性低。
true
在ISR列表中没有副本的情况下，可以选择任意一个没有宕机的主机作为新的leader，该策略可用性高，但可靠性没有保证。
当Consumer由于消费能力低而引发了消费超时，则可能会形成重复消费。
在某数据刚好消费完毕，但是正准备提交offset时候，消费时间超时，则broker认为这条消息未消费成功。这时就会产生重复消费问题。
其解决方案：延长offset提交时间。
当Consumer消费了消息，但还没有提交offset时宕机，则这些已经被消费过的消息会被重复消费。
其解决方案：将自动提交改为手动提交。
其实在开发的时候, 我们在设计程序的时候, 比如考虑到网络故障等一些异常的情况, 我们都会设置消息的重试次数,
可能还有其他可能出现消息重复, 那我们应该如何解决呢?
下面提供三个方案:
给每个消息都设置一个独一无二的uuid, 所有的消息, 我们都要存一个uuid, 我们在消费消息的时候, 首先去持久化系统中查询一下, 看这个看是否以前消费过, 如没有消费过, 在进行消费, 如果已经消费过, 丢弃就好了, 下图, 表明了这种方案
幂等（Idempotence）在数学上是这样定义的，如果一个函数 f(x) 满足：f(f(x)) = f(x)，则函数 f(x) 满足幂等性。
这个概念被拓展到计算机领域，被用来描述一个操作、方法或者服务。一个幂等操作的特点是，其任意多次执行所产生的影响均与一次执行的影响相同。一个幂等的方法，使用同样的参数，对它进行多次调用和一次调用，对系统产生的影响是一样的。所以，对于幂等的方法，不用担心重复执行会对系统造成任何改变。
我们举个例子
标签：数据同步 rom 重试 leo ble 流处理 文件系统 行数据 发送
原文地址：https://blog.51cto.com/lingjing/2476483
