主题信息（必填）
主题描述（最多限制在50个字符）
申请人信息（必填）
申请信息已提交审核，请注意查收邮件，我们会尽快给您反馈。
如有疑问，请联系
开发者们的力量是无穷滴~欢迎一线的移动开发者们撰文分享你们的进阶、踩过的坑以及对热点技术的看法。一起关注 iOS、Android、跨平台、IoT、VR/AR等技术实践，投稿邮箱：。
与世界分享我的认识、见解，我是it人，我喜欢it
既然选择了ACM了,那就好好爱它,不管风雨兼程.
从业10年＋的IT媒体人，再次回归组织。
Android路上，一起爬坑。。。消息中间件-Kafka（49）
LinkedIn在2011年7月开始大规模使用Kafka，当时Kafka每天大约处理10亿条消息，这一数据在2012年达到了每天200亿条，而到了2013年7月，每天处理的消息达到了2000亿条。在几个月前，他们的最新记录是每天利用Kafka处理的消息超过1万亿条，在峰值时每秒钟会发布超过450万条消息，每周处理的信息是1.34 PB。每条消息平均会被4个应用处理。在过去的四年中，实现了1200倍的增长。
随着规模的不断扩大，LinkedIn更加关注于Kafka的可靠性、成本、安全性、可用性以及其他的基础指标。在这个过程中，LinkedIn的技术团队在多个特性和领域都进行了有意义的探索。
Linkedln在Kafka上的主要关注领域
配额（Quotas）
在LinkedIn，不同的应用使用同一个Kafka集群，所以如果某个应用滥用Kafka的话，将会对共享集群的其他应用带来性能和SLA上的负面影响。有些合理的使用场景有可能也会带来很坏的影响，比如如果要重新处理整个数据库的所有数据的话，那数据库中的所有记录会迅速推送到Kafka上，即便Kafka性能很高，也会很容易地造成网络饱和和磁盘冲击。
Kartik Paramasivam绘图展现了不同的应用是如何共享Kafka Broker的：
为了解决这个问题，LinkedIn的团队研发了一项特性，如果每秒钟的字节数超过了一个阈值，就会降低这些Producer和Customer的速度。对于大多数的应用来讲，这个默认的阈值都是可行的。但是有些用户会要求更高的带宽，于是他们引入了白名单机制，白名单中的用户能够使用更高数量的带宽。这种配置的变化不会对Kafka Broker的稳定性产生影响。这项特性运行良好，在下一版本的Kafka发布版中，所有的人就都能使用该特性了。
开发新的Customer
目前的Kafka Customer客户端依赖于ZooKeeper，这种依赖会产生一些大家所熟知的问题，包括ZooKeeper的使用缺乏安全性以及Customer实例之间可能会出现的脑裂现象（split brain）。因此，LinkedIn与Confluent以及其他的开源社区合作开发了一个新的Customer。这个新的Customer只依赖于Kafka Broker，不再依赖于ZooKeeper。这是一项很复杂的特性，因此需要很长的时间才能完全应用于生产环境中。
在Kafka中，目前有两个不同类型的Customer。如果Customer希望完全控制使用哪个分区上的Topic的话，就要使用低级别的Customer。在高级别的Customer中，Kafka客户端会自动计算如何在Customer实例之间分配Topic分区。这里的问题在于，如果使用低级别Customer的话，会有许多的基本任务要去完成，比如错误处理、重试等等，并且无法使用高级别Customer中的一些特性。在LinkedIn这个新的Customer中，对低级别和高级别的Customer进行了调和。
可靠性和可用性的提升
按照LinkedIn这样的规模，如果Kafka的新版本中有什么重要缺陷的话，就会对可靠性产生很大的影响。因此，LinkedIn技术团队一项很重要的任务就是发现和修正缺陷。他们在可靠性方面所做的增强包括：
Mirror Maker无损的数据传输：Mirror Maker是Kafka的一个组件，用来实现Kafka集群和Kafka Topic之间的数据转移。LinkedIn广泛使用了这项技术，但是它在设计的时候存在一个缺陷，在传输时可能会丢失数据，尤其是在集群升级或机器重启的时候。为了保证所有的消息都能正常传输，他们修改了设计，能够确保只有消息成功到达目标Topic时，才会认为已经完全消费掉了。
副本的延迟监控：所有发布到Kafka上的消息都会复制副本，以提高持久性。当副本无法“跟上”主版本（master）的话，就认为这个副本处于非健康的状态。在这里，“跟上”的标准指的是配置好的字节数延迟。这里的问题在于，如果发送内容很大的消息或消息数量不断增长的话，那么延迟可能会增加，那么系统就会认为副本是非健康的。为了解决这个问题，LinkedIn将副本延迟的规则修改为基于时间进行判断。
实现新的Producer：LinkedIn为Kafka实现了新的Producer，这个新的Producer允许将消息实现为管道（pipeline），以提升性能。目前该功能尚有部分缺陷，正在处于修复之中。
删除Topic：作为如此成熟的产品，Kafka在删除Topic的时候，会出现难以预料的后果或集群不稳定性，这一点颇令人惊讶。在几个月前，LinkedIn对其进行了广泛地测试并修改了很多缺陷。到Kafka的下一个主版本时，就能安全地删除Topic了。
在Kafka中，这是参与者最多的特性之一，众多的公司互相协作来解决这一问题。其成果就是加密、认证和权限等功能将会添加到Kafka中，在LinkedIn，预期在2015年能使用加密功能，在2016年能使用其他的安全特性。
Kafka监控框架
LinkedIn最近正在致力于以一种标准的方式监控Kafka集群，他们的想法是运行一组测试应用，这些应用会发布和消费Kafka Topic数据，从而验证基本的功能（顺序、保证送达和数据完整性等）以及端到端发布和消费消息的延时。除此之外，这个框架还可以验证Kafka新版本是否可以用于生产环境，能够确保新版本的Kafka Broker不会破坏已有的客户端。
当拿到新的Kafka开源版本后，LinkedIn会运行一些故障测试，从而验证发生失败时Kafka新版本的质量。针对这项任务，LinkedIn研发了名为Simoorg的故障引导框架，它会产生一些低级别的机器故障，如磁盘写失败、关机、杀进程等等。
应用延迟监控
LinkedIn开发了名为Burrow的工具，能够监控Customer消费消息的延迟，从而监控应用的健康状况。
保持Kafka集群平衡
LinkedIn在如下几个维度保证了集群的平衡：
感知机柜：在进行平衡时，很重要的一点是Kafka分区的主版本与副本不要放到同一个数据中心机柜上。如果不这样做的话，一旦出现机柜故障，将会导致所有的分区不可用。
确保Topic的分区公平地分发到Broker上：在为Kafka发布和消费消息确定了配额后，这项功能变得尤为重要。相对于将Topic的分区发布到同一个Broker节点上，如果Topic的分区能够均衡地分发到多个Broker上，那么相当的它有了更多的带宽。
确保集群节点的磁盘和网络容量不会被耗尽：如果几个Topic的大量分区集中到了集群上的少数几个节点上，那么很容易出现磁盘或网络容量耗尽的情况。
在LinkedIn，目前维护站点可靠性的工程师（Site Reliability Enginee，SRE）通过定期转移分区确保集群的平衡。在分区放置和重平衡方面，他们已经做了一些原始设计和原型实现，希望能够让系统更加智能。
在其他的数据系统中，将Kafka作为核心的组成部分
在LinkedIn，使用Espresso作为NoSQL数据库，目前他们正在将Kafka作为Espresso的备份机制。这将Kafka放到了站点延迟敏感数据路径的关键部分，同时还需要保证更高的消息传送可靠性。目前，他们做了很多的性能优化，保证消息传输的低延迟，并且不会影响消息传递的可靠性。
Kafka还会用于异步上传数据到Venice之中。除此之外，Kafka是Apache Samza实时流处理的一个重要事件源，同时Samza还使用Kafka作为持久化存储，保存应用的状态。在这个过程中，LinkedIn修改了一些重要的缺陷，并增强了Kafka的日志压缩特性。
LinkedIn的Kafka生态系统
除了Apache Kafka Broker、客户端以及Mirror Maker组件之外，LinkedIn还有一些内部服务，实现通用的消息功能：
支持非Java客户端：在LinkedIn，会有一些非Java应用会用到Kafka的REST接口，去年他们重新设计了Kafka的REST服务，因为原始的设计中并不能保证消息的送达。
消息的模式：在LinkedIn，有一个成熟的“模式（schema）注册服务”，当应用发送消息到Kafka中的时候，LinkedIn Kafka客户端会根据消息注册一个模式（如果还没有注册过的话）。这个模式将会自动在Customer端用于消息的反序列化。
成本计算：为了统计各个应用对Kafka的使用成本，LinkedIn使用了一个Kafka审计Topic。LinkedIn客户端会自动将使用情况发送到这个Topic上，供Kafka审计服务读取并记录使用情况，便于后续的分析。
审计系统：LinkedIn的离线报告job会反映每小时和每天的事件情况，而事件从源Kafka Topic/集群/数据中心，到最后的HDFS存储是需要时间的。因此，Hadoop job需要有一种机制，保证某个时间窗口能够获得所有的事件。LinkedIn Kafka客户端会生成它们所发布和消费的消息数量。审计服务会记录这个信息，Hadoop以及其他的服务可以通过REST接口获取这一信息。
支持内容较大的消息：在LinkedIn，将消息的大小限定为1MB，但是有些场景下，无法满足这一限制。如果消息的发布方和使用方是同一个应用的话，一般会将消息拆分为片段来处理。对于其他的应用，建议消息不要超过1MB。如果实在无法满足该规则的话，消息的发送和消费方就需要使用一些通用的API来分割和组装消息片段，而在LinkedIn的客户端SDK中，他们实现了一种特性，能够自动将一条大的信息进行分割和重组。
目前，越来越多的国内外公司在使用Kafka，如Yahoo!、Twitter、Netflix和Uber等，所涉及的功能从数据分析到流处理不一而足，希望LinkedIn的经验也能够给其他公司一些借鉴。
来源：InfoQ
转载请注明来自36大数据（)：&>>&
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：405141次
积分：5957
积分：5957
排名：第3769名
原创：21篇
转载：1142篇
评论：19条
(1)(1)(56)(43)(43)(47)(63)(29)(51)(124)(23)(25)(10)(27)(20)(17)(14)(1)(42)(41)(2)(20)(38)(232)(4)(3)(28)(4)(43)(20)(22)(51)(1)(21)(1)(1)(6)(1)(1)(1)114网址导航KAFKA：如何做到1秒发布百万级条消息-提供留学,移民,理财,培训,美容,整形,高考,外汇,印刷,健康,建材等信息_突袭网
当前位置&:&&&&KAFKA：如何做到1秒发布百万级条消息
热门标签：&
KAFKA：如何做到1秒发布百万级条消息
来源： 由用户
编辑：王亮
KAFKA是分布式发布-订阅消息系统，是一个分布式的，可划分的，冗余备份的持久性的日志服务。它主要用于处理活跃的流式数据。现在被广泛地应用于构建实时数据管道和流应用的场景中，具有横向扩展，容错，快等优点，并已经运行在众多大中型公司的生产环境中，成功应用于大数据领域，本文分享一下我所了解的KAFKA。【KAFKA高吞吐率性能揭秘】KAFKA的第一个突出特定就是“快”，而且是那种变态的“快”，在普通廉价的虚拟机器上，比如一般SAS盘做的虚拟机上，据LINDEDIN统计，最新的数据是每天利用KAFKA处理的消息超过1万亿条，在峰值时每秒钟会发布超过百万条消息，就算是在内存和CPU都不高的情况下，Kafka的速度最高可以达到每秒十万条数据，并且还能持久化存储。作为消息队列，要承接读跟写两块的功能，首先是写，就是消息日志写入KAFKA，那么，KAFKA在“写”上是怎么做到写变态快呢？首先，可以使用KAFKA提供的生产端API发布消息到1个或多个Topic（主题）的一个（保证数据的顺序）或者多个分区（并行处理，但不一定保证数据顺序）。Topic可以简单理解成一个数据类别，是用来区分不同数据的。KAFKA维护一个Topic中的分区log，以顺序追加的方式向各个分区中写入消息，每个分区都是不可变的消息队列。分区中的消息都是以k-v形式存在。k表示offset，称之为偏移量，一个64位整型的唯一标识，offset代表了Topic分区中所有消息流中该消息的起始字节位置。v就是实际的消息内容，每个分区中的每个offset都是唯一存在的，所有分区的消息都是一次写入，在消息未过期之前都可以调整offset来实现多次读取。以上提到KAFKA“快”的第一个因素：消息顺序写入磁盘。我们知道现在的磁盘大多数都还是机械结构（SSD不在讨论的范围内），如果将消息以随机写的方式存入磁盘，就会按柱面、磁头、扇区的方式进行（寻址过程），缓慢的机械运动（相对内存）会消耗大量时间，导致磁盘的写入速度只能达到内存写入速度的几百万分之一，为了规避随机写带来的时间消耗，KAFKA采取顺序写的方式存储数据，如下图所示：新来的消息只能追加到已有消息的末尾，并且已经生产的消息不支持随机删除以及随机访问，但是消费者可以通过重置offset的方式来访问已经消费过的数据。 即使顺序读写，过于频繁的大量小I/O操作一样会造成磁盘的瓶颈，所以KAFKA在此处的处理是把这些消息集合在一起批量发送，这样减少对磁盘IO的过度读写，而不是一次发送单个消息。 另一个是无效率的字节复制，尤其是在负载比较高的情况下影响是显着的。为了避免这种情况，KAFKA采用由Producer，broker和consumer共享的标准化二进制消息格式，这样数据块就可以在它们之间自由传输，无需转换，降低了字节复制的成本开销。 同时，KAFKA采用了MMAP(Memory Mapped Files，内存映射文件)技术。很多现代操作系统都大量使用主存做磁盘缓存，一个现代操作系统可以将内存中的所有剩余空间用作磁盘缓存，而当内存回收的时候几乎没有性能损失。 由于KAFKA是基于JVM的，并且任何与Java内存使用打过交道的人都知道两件事： ? 对象的内存开销非常高，通常是实际要存储数据大小的两倍； ? 随着数据的增加，java的垃圾收集也会越来越频繁并且缓慢。 基于此，使用文件系统，同时依赖页面缓存就比使用其他数据结构和维护内存缓存更有吸引力： ? 不使用进程内缓存，就腾出了内存空间，可以用来存放页面缓存的空间几乎可以翻倍。 ? 如果KAFKA重启，进行内缓存就会丢失，但是使用操作系统的页面缓存依然可以继续使用。 可能有人会问KAFKA如此频繁利用页面缓存，如果内存大小不够了怎么办？ KAFKA会将数据写入到持久化日志中而不是刷新到磁盘。实际上它只是转移到了内核的页面缓存。 利用文件系统并且依靠页缓存比维护一个内存缓存或者其他结构要好，它可以直接利用操作系统的页缓存来实现文件到物理内存的直接映射。完成映射之后对物理内存的操作在适当时候会被同步到硬盘上。 KAFKA除了接收数据时写得快，另外一个特点就是推送数据时发得快。 KAFKA这种消息队列在生产端和消费端分别采取的push和pull的方式，也就是你生产端可以认为KAFKA是个无底洞，有多少数据可以使劲往里面推送，消费端则是根据自己的消费能力，需要多少数据，你自己过来KAFKA这里拉取，KAFKA能保证只要这里有数据，消费端需要多少，都尽可以自己过来拿。▲零拷贝 具体到消息的落地保存，broker维护的消息日志本身就是文件的目录，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。维护这个公共的格式并允许优化最重要的操作：网络传输持久性日志块。 现代的unix操作系统提供一个优化的代码路径，用于将数据从页缓存传输到socket；在Linux中，是通过sendfile系统调用来完成的。Java提供了访问这个系统调用的方法：FileChannel.transferTo API。 要理解senfile的影响，重要的是要了解将数据从文件传输到socket的公共数据路径，如下图所示，数据从磁盘传输到socket要经过以下几个步骤：? 操作系统将数据从磁盘读入到内核空间的页缓存 ? 应用程序将数据从内核空间读入到用户空间缓存中 ? 应用程序将数据写回到内核空间到socket缓存中 ? 操作系统将数据从socket缓冲区复制到网卡缓冲区，以便将数据经网络发出 这里有四次拷贝，两次系统调用，这是非常低效的做法。如果使用sendfile，只需要一次拷贝就行：允许操作系统将数据直接从页缓存发送到网络上。所以在这个优化的路径中，只有最后一步将数据拷贝到网卡缓存中是需要的。常规文件传输和zeroCopy方式的性能对比：假设一个Topic有多个消费者的情况， 并使用上面的零拷贝优化，数据被复制到页缓存中一次，并在每个消费上重复使用，而不是存储在存储器中，也不在每次读取时复制到用户空间。 这使得以接近网络连接限制的速度消费消息。 这种页缓存和sendfile组合，意味着KAFKA集群的消费者大多数都完全从缓存消费消息，而磁盘没有任何读取活动。▲批量压缩 在很多情况下，系统的瓶颈不是CPU或磁盘，而是网络带宽，对于需要在广域网上的数据中心之间发送消息的数据流水线尤其如此。所以数据压缩就很重要。可以每个消息都压缩，但是压缩率相对很低。所以KAFKA使用了批量压缩，即将多个消息一起压缩而不是单个消息压缩。 KAFKA允许使用递归的消息集合，批量的消息可以通过压缩的形式传输并且在日志中也可以保持压缩格式，直到被消费者解压缩。 KAFKA支持Gzip和Snappy压缩协议。【KAFKA数据可靠性深度解读】KAFKA的消息保存在Topic中，Topic可分为多个分区，为保证数据的安全性，每个分区又有多个Replia。? 多分区的设计的特点：1.为了并发读写，加快读写速度；2.是利用多分区的存储，利于数据的均衡；3.是为了加快数据的恢复速率，一但某台机器挂了，整个集群只需要恢复一部分数据，可加快故障恢复的时间。每个Partition分为多个Segment，每个Segment有.log和.index 两个文件，每个log文件承载具体的数据，每条消息都有一个递增的offset，Index文件是对log文件的索引，Consumer查找offset时使用的是二分法根据文件名去定位到哪个Segment，然后解析msg，匹配到对应的offset的msg。 每个Partition会在磁盘记录一个RecoveryPoint,，记录已经flush到磁盘的最大offset。当broker 失败重启时，会进行loadLogs。首先会读取该Partition的RecoveryPoint，找到包含RecoveryPoint的segment及以后的segment， 这些segment就是可能没有完全flush到磁盘segments。然后调用segment的recover，重新读取各个segment的msg，并重建索引。每次重启KAFKA的broker时，都可以在输出的日志看到重建各个索引的过程。Producer和Consumer都只与Leader交互，每个Follower从Leader拉取数据进行同步。如上图所示，ISR是所有不落后的replica集合，不落后有两层含义：距离上次FetchRequest的时间不大于某一个值或落后的消息数不大于某一个值，Leader失败后会从ISR中随机选取一个Follower做Leader，该过程对用户是透明的。 当Producer向Broker发送数据时,可以通过request.required.acks参数设置数据可靠性的级别。 此配置是表明当一次Producer请求被认为完成时的确认值。特别是，多少个其他brokers必须已经提交了数据到它们的log并且向它们的Leader确认了这些信息。?典型的值：0： 表示Producer从来不等待来自broker的确认信息。这个选择提供了最小的时延但同时风险最大（因为当server宕机时，数据将会丢失）。1：表示获得Leader replica已经接收了数据的确认信息。这个选择时延较小同时确保了server确认接收成功。-1：Producer会获得所有同步replicas都收到数据的确认。同时时延最大，然而，这种方式并没有完全消除丢失消息的风险，因为同步replicas的数量可能是1。如果你想确保某些replicas接收到数据，那么你应该在Topic-level设置中选项min.insync.replicas设置一下。 仅设置 acks= -1 也不能保证数据不丢失,当ISR列表中只有Leader时,同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1，还要保证ISR的大小大于等于2。?具体参数设置： request.required.acks:设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功。 min.insync.replicas: 设置为>=2,保证ISR中至少两个Replica。 Producer：要在吞吐率和数据可靠性之间做一个权衡。 KAFKA作为现代消息中间件中的佼佼者，以其速度和高可靠性赢得了广大市场和用户青睐，其中的很多设计理念都是非常值得我们学习的，本文所介绍的也只是冰山一角，希望能够对大家了解KAFKA有一定的作用。
更多精彩 >>>