kafka创建主题 kafka创建主题超时
kafka架构详解
以上内容参考：kafka是一个高性能、低延迟的分布式发布-消息系统，适用于在线、离线消息消费，为了防止数据丢失，kafka将消息持久化到磁盘上并在集群内.
kafka创建主题 kafka创建主题超时
kafka创建主题 kafka创建主题超时
Topic
在深入了解kafka之前，先介绍kafka体系架构中的一些组件，包括Topic、Producer、Consumer、Consumer Group、Broker、Partition、Leader、Follower。
消息被发送到kafak中都有一个类别，这个类别叫做Topic，kafka中的消息都是通过主题进行组织的，一个Topic可以有1个或多个Partition。
Producer
生产者，即是将消息发送到kafka的过程，发送消息是需要指定Topic，并且可以指定Partition。Broker接收到消息后，将消息存放在partition中。
Consumer
消费者，从broker topic中读取消息，可以一个或多个topic。
消费者组由一个或多个消费者组成，消费者组中的消费者共同消费一个主题的分区，主题中每个分区只能被同一个消费者组中的一个消费者消费。
Broker
kafka集群包括一个或多个，每个就叫做Broker。
Partition
Topic中的数据可以分割为一个或多个Partition，Partition在底层逻辑是log文件，每个Partition由多个Segment组成，任何发送到kafka的消息最终都是会被追加到log文件的尾部。
Leader
Topic的Partition允许有若干个副本，每个分区都一个leader和多个follower，leader负责所有的读写作。
Follower追随Leader，所有的读写请求都是通过Leader路由，请求会广播给所有的Follower，Follower和Leader保持数据同步。如果Leader失效，通过Follower选举一个新的Leader.
下面通过一张简单的UML图简要说明组件之间的交互和关联关系
- kafka集群可以有1个或多个Broker
- Broker 可以包含多个副本（每个分区可以包含多个副本，通常每个分区副本数不会多于Broker数量，一个broker中包含很多个分区）
- Topic可以有1个或多个分区
- broker中的每个partition可以有0个或1个副本
- 一个partition有一个leader副本和0个或多个follower副本
- 每个partition副本位于一个broker上，并且一个partition不能划分多个broker。
Kafka架构
下面重点介绍Producer、Topic、Consumer的关系，一个简单生产消费的过程例子如下图所示
在这个例子中，一个生产者将消息发送给单个topic
上面这个图中，1个生产者发布消息到1个topic，一个消费者消费1个Topic，如上图中的Producer 1和Consumer 1；一个Topic可以是由多个生产者发布消息，如Topic4；1个消费者可以消费多个Topic，如图中的Consumer 2。
如上图的例子，一个生产者可以给多个Topic发布消息。一个消费者同一时间只能给一个topic发布消息，但是可以使用异步发布消息，生产者可以一次将多个消息发送给多个Topic.
生产者负责将每条消息发送到分区，默认分区由消息key通过hash算法确定，如果没有指定消息key，则通过循环轮询来确定分区。但是在实际业务场景中，默认的分区行为并不能满足业务需要，比如需要确保消息的顺序或需要将消息平均分配给消费者等等。因此，生产者在发布消息的时候可以使用自定义分区方式，为消息指定分区key、重写分区算法或手动设置分区等方式将消息发布到特定分区。
kafka内部运作的基本逻辑大概为：每个主题都有1个或多个分区，这些分区分不在1个或多个Broker上，为了提高消息的可靠性不会丢失，可以配置多个副本因子，这样每个分区可以被到一个或多个Broker上，每个分区对应一个log文件，只能被一个消费组中的一个消费者消费，用于提高Topic的并发性。因此一般将消费组消费者数量设置为小于或者等于topic的分区数量，如果要增加消费者也相应的增加对应的分区数量。
同一个分区内的消息是由顺序的，每个分区仅能被同一个消费组中的一个消费者顺序消费。但是，当多个消费组了同一个topic，则每个组中的消费者都会收到消息。
下面例子说明多分区情况下，消费者组和消费者消费的几种情况。
分区数和消费者数相同，如下图所示
这种情况，同一个消费组的每个消费者只消费一个分区。
消费者数量多于分区数，则某些消费者就处于空闲状态，当有消费者down掉或添加新的分区情况时，空闲消费者将发挥作用。
另外一种情况，消费者数比分区数少，如下图所示
这种情况，导致某些分区需要负责更多的分区，处理更多的消息。
，多个消费组消费了同一个topic
topic消息被广播到每个消费组，每个消费组都可以接受同一个消息。这是kafka实现一个Topic广播的方式，一个Topic可以被多个Conumse Group的消费者同时消费；同一个消息只能被一个消费者组中的一个消费者消费。
Kafka 技术汇总
首先你在链接时候检查是否代码里的IP 和端口是不是对的，端口是broker 端口，默认909消息系统由生产者 存储系统和消费者组成 章分析了生产者发送消息给服务端的过程，本章分析消费者从服务端存储系统读取生产者写入消息的过程 首先我 来了解消费者的 些基础知识2 ；Kafka 是一个分布式消息队列，具有高性能、持久化、多副本备份、横向扩展能力。生产者往队列里写消息，消费者从队列里取消息进行业务逻辑。一般在架构设计中起到解耦、削峰、异步处理的作用。
kafka系列之(3)——Coordinator与offset管理和Consumer Rebalance
4.上面提到了kafka的ISR机制，kafka的容错性就是由ISR的机制来保证的。consumer group是kafka提供的可扩展且具有容错性的消费者机制。组内可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费主题(subscribed topics)的所有分区(partition)。
consumer group下可以有一个或多个consumer instance，consumer instance可以是一个进程，也可以是一个线程
group.id是一个字符串，标识一个consumer group
consumer group下的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)
Coordinator一般指的是运行在broker上的group Coordinator，用于管理Consumer Group中各个成员，每个Kafka都有一个GroupCoordinator实例，管理多个消费者组，主要用于offset位移管理和Consumer Rebalance。
consumer group如何确定自己的coordinator是谁呢？ 简单来说分为两步：
消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在Kafka中这个位置信息有个专门的术语：位移(offset)。
(1)、很多消息引擎都把这部分信息保存在端(broker端)。这样做的好处当然是实现简单，但会有三个主要的问题：
1. broker从此变成有状态的，会影响伸缩性；
2. 需要引入应答机制(acknowledgement)来确认消费成功。
3. 由于要保存很多consumer的offset信息，必然引入复杂的数据结构，造成资源浪费。
而Kafka选择了不同的方式：每个consumer group管理自己的位移信息，那么只需要简单的一个整数表示位置就够了；同时可以引入checkpoint机制定期持久化，简化了应答机制的实现。
(2)、Kafka默认是定期帮你自动提交位移的(enable.automit = true)，你当然可以选择手动提交位移实现自己控制。
(3)、另外kafka会定期把group消费情况保存起来，做成一个offset map，如下图所kafka在2010年开源，2011年7月正式进入Apache进行孵化，2012年10月顺利毕业，后成为Apache的项目。示：
上图中表明了test-group这个组当前的消费情况。
老版本的位移是提交到zookeeper中的，目录结构是：/consumers/ ，但是zookeeper其实并不适合进行大批量的读写作，尤其是写作。 __consumers_offsets topic配置了compact策略，使得它总是能够保存的位移信息，既控制了该topic总体的日志容量，也能实现保存offset的目的。compact的具体原理请参见： Log Compaction 至于每个group保存到__consumers_offsets的哪个分区，如何查看的问题请参见这篇文章： Kafka 如何读取offset topic内容 (__consumer_offsets) offset提交消息会根据消费组的key(消费组名称)进行分区. 对于一个给定的消费组,它的所有消息都会发送到的broker(即Coordinator) Coordinator上负责管理offset的组件是 Offset mar 。负责存储,抓取,和维护消费者的offsets. 每个broker都有一个offset mar实例. 有两种具体的实现: DefaultOffsetMar: 提供消费者offsets内置的offset管理。 通过在config/server.properties中的offset.storage参数选择。 DefaultOffsetMar 除了将offset作为logs保存到磁盘上,DefaultOffsetMar维护了一张能快速服务于offset抓取请求的 consumer offsets表 。这个表作为缓存，包含的含仅仅是”offsets topic”的partitions中属于leader partition对应的条目(存储的是offset)。 对于DefaultOffsetMar还有两个其他属性: “和””,默认值都是1。这两个属性会用来自动地创建”offsets topic”。 offset mar接口的概要： 什么是rebalance？ rebalance本质上是一种协议，规定了一个consumer group下的所有consumer如何达成一致来分配topic的每个分区。比如某个group下有20个consumer，它了一个具有100个分区的topic。正常情况下，Kafka平均会为每个consumer分配5个分区。这个分配的过程就叫rebalance。Kafka新版本consumer默认提供了两种分配策略：range和round-robin。 rebalance的触发条件有三种： 组成员发生变更(新consumer加入组、已有consumer主动离开组或已有consumer崩溃了——这两者的区别后面会谈到) 主题数发生变更——这当然是可能的，如果你使用了正则表达式的方式进行，那么新建匹配正则表达式的topic就会触发rebalance 主题的分区数发生变更 refer 其次查看代码是生产者，看Kafka 集群里这个主题是否存在（如果不存在，默认是配置可以自动创建，看是非将该配置修改）；然后检测防火墙，相应端口是否开放（防火墙直接关也可以）；检测 server.properties 文件的 listeners 是否配置，若没有将其配置好 一、kafka定义 二、kafka的优势 三、kafka的原理 一、Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为开源项目。 二、kafka的优势 高吞吐量、低延迟：kafka美妙之处是可以处理几十万条信息，它的延迟只有几毫秒，每个topic可以分多个partition，consumer group对partition进行consume作。 可扩展性：kafka集群支持热扩展 持久化、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失 容错性：允许集群中失败（若副本数量为n，则允许n-1个失败） 高并发：支持数千个客户端同时读写 三、kafka的原理 kafka是如何实现以上所述这几点，我们逐一说明： 1.高吞吐量、低延迟 kafka采取了一定量的批处理机制，即当生产数据达到一定数量或者达到时间窗口后，将所收集到的数据一批次的提交到，我们设处理一次数据的时间为1ms，那每秒钟能处理1000条，延时为1ms，如果此时将处理间隔变成9ms，即每10ms处理一批数据，设这段时间接收到100条处理，那每秒则能处理10000条，但是延时变成了10ms。为了获得的吞吐量，需要牺牲一定的延迟，但是这样的牺牲是值得的。当确定了这种小批量方式之后，高速的写则取决于kafka自身写磁盘的速度了。而由于kafka本身对数据不做任何的处理，只管写入数据，保管数据，分发数据，因此会是一种批量顺序写入数据的情况，而磁盘的读写速度大量消耗在寻址上，也就是随机读写，但是对于顺序写入的速度是非常快的，甚至能媲美内存的随机写入速度。有人做过一个对比，普通磁盘顺序写入每秒能达到53.2M/s，SSD的顺序写入速度为42.2M/s，内存的顺序写入速度为358.2M/s。kafka正是利用了这个特性，顺序写入，速度相对较快。而kafka本身虽然也是写入磁盘持久化数据，但实际上kafka是将数据顺序写入页缓存中（page cache），然后由作系统自行决定何时写到磁盘上，因此kafka的写作能在每秒轻轻松松达到写入数十万条记录。并且基于kafka的动态扩展，这个数字还能不断增大。 kafka在消费端也有着高吞吐量，由于kafka是将数据写入到页缓存中，同时由于读写相间的间隔并不大，很大可能性会在缓存中命中，从而保证高吞吐量。另外kafka由于本身不对数据做任何的修改，完全使用零拷贝技术，大大提升数据的读取能力。 2.kafka每个叫做broker，而每一个broker都是运行的，可以随时加入kafka集群，集群的心跳管理是由zookeeper负责，新加入的broker只要broker id不与原有的冲突就能顺利的加入集群中，实现动态扩展。 3.kafka的持久化在上面已经提到，kafka绕过了ja的堆处理数据，直接将数据写入页缓存，然后由作系统来管理页缓存写入磁盘，实现持久化。kafka每一个主题topic是一个业务数据，他可由多个partition组成，而每个partition可以有多个replica副本，用于保证数据的可靠性。replica分为两个角色，一个是leader，一个是追随者，同一时间，每一个partition只能有一个leader，其他都是追问随者，laeder负责接收数据并写入log，而追随者不能被用户写入数据，只是从leader角色的replica副本中同步log写入自己的log，保持数据同步。kafka中有一个概念，ISR，全称是in-sync replica，即所有可用的replica副本，这里的ISR数量只要大于1，这个partition就能正常运作，因此容错性非常好，设n个replica，那最多可以坏n-1个replica的情况下，还能保持系统正常运行。当replica迟滞到一定时间后，会被kafka从ISR中剔除，当再次同步后，可以再次加入ISR，如果这时候leader出现问题，会从ISR中重新选举一个leader，原先的leader再次同步成功后会重新加入ISR，成为一个flower。 四、kafka的起源 kafka起源于LinkedIn公司，当时领英公司需要收集两大类数据，一是业务系统和应用程序的性能指标数据，而是用户的作行为数据。当时为了收集这两类数据，领英自研了两套相应的数据收集系统，但是这两套系统都存在一些弊端，无法实现实时交互、实时性、维护成本高。因此领英的工程师希望找到一个统一的组件来收集分发消费这些大批量的数据，ActiveMQ由于扩展性不足，不能支撑大数据量而被抛弃，从而决定自研一套满足需求的系统组件，也就是kafka。 kafka的设计之初主要有三个目标： 1.为生产者和消费者提供一套简单的API 2.降低网络传输和磁盘存储开销 3.具有高伸缩性架构 目前kafka可以算是超额完成了目标。 kafka的名称由来也很有意思，因为kafka系统的写作性能特别强，因此想使用一个作家的名字来命名kafka，而Jay Kreps，kafka的三位作者之一，在上大学的时候很喜欢Franz Kafka，因此起来这样一个名字。 若没有分区，一个topic对应的消息集在分布式集群服务组中，就会分布不均匀，即可能导致某台A记录当前topic的消息集很多，若此topic的消息压力很大的情况下，A就可能导致压力很大，吞吐也容易导致瓶颈。 有了分区后，设一个topic可能分为10个分区，kafka内部会根据一定的算法把10分区尽可能均匀分布到不同的上，比如：A负责topic的分区1，B负责topic的分区2，在此情况下，Producer发消息时若没指定发送到哪个分区的时候，kafka就会根据一定算法上个消息可能分区1，下个消息可能在分区2。当然高级API也能自己实现其分发算法。 partitions的设计目的有多个.最根本原因是kafka基于文件存储.通过分区,可以将日志内容分散到多个server上,来避免文件尺寸达到单机磁盘的上限,每个partiton都会被当前server(kafka实例)保存;可以将一个topic切分多任意多个partitions,来消息保存/消费的效率.此外越多的partitions意味着可以容纳更多的consumer,有效提升并发消费的能力 kafka topic 分区的原因主要是为了并发访问，增大吞吐量。 具体来讲，每一个topic均被分成多个互不重复的partition，其中每个partition均被放在一个或多个broker上，这样就提供了容错。如果数据在不同的patition上，那么他们的访问将可能由不同的broker来完成，这样就实现了并发访问。分区的信息储存在Zookeeper上。 所以，分区是kafka用来提供并发访问控制的一个概念。 Kafka可以将主题划分为多个分区，根据分区规则会把不同的消息存储在不同的分区上，只要分区规则设计的合理，那么所有的消息将均匀地分布在不同的分区上，这样就实现了负载均衡和 消息的水平扩展。 Kafka作为一个传统的消息的替代品表现得非常出色。使用消息有各种各样的原因(将处理与数据生成器解耦，缓冲未处理的消息，等等)。与大多数消息传递系统相比，Kafka有更好的吞吐量、内置分区、和容错性，这使得它成为大规模消息处理应用的一个很好的解决方案。 根据我们如果是协调者发生故障，服务端会有自己的故障容错机制，选出管理消费组所有消费者的新协调者节，点消费者客户端没有权利做这个工作，它能做的只是等待一段时间，查询服务端是否已经选出了新的协调如果消费者查到现在已经有管理协调者的协调，就会连接这个新协调节，哉由于这个协调是服务端新选出来的，所以每个消费者都应该重新连接协调的经验，消息传递的使用通常是相对较低的吞吐量，但可能需要较低的端到端延迟，并且常常依赖于Kafka提供的强大的持久性保证。 在这个领域，Kafka可以与ActiveMQ或RabbitMQ等传统消息传递系统相媲美。 活动跟踪通常是非常大的量，因为许多活动消息会生成的每个用户页面视图。 Kafka通常用于运行数据。这涉及聚合来自分布式应用程序的统计信息，以生成集中的作数据提要。 许多人使用Kafka作为日志聚合解决方案的替代品。日志聚合通常收集上的物理日志文件，并将它们放在一个中心位置(可能是文件或HDFS)进行处理。Kafka抽象了文件的细节，并以消息流的形式对日志或数据进行了更清晰的抽象。这允许低延迟处理，并更容易支持多个数据源和分布式数据消费。与以日志为中心的系统如Scribe或Flume相比，Kafka提供了同样好的性能，由于而更强的持久性保证，以及更低的端到端延迟。 很多Kafka的用户在处理数据的管道中都有多个阶段，原始的输入数据会从Kafka的主题中被消费，然后被聚合、充实或者转换成新的主题进行进一步的消费或者后续的处理。例如，文章的处理管道可能会从RSS源抓取文章内容，并将其发布到“文章”主题;进一步的处理可能会规范化或删除该内容，并将清理后的文章内容发布到新主题;一个处理阶段可能会尝试向用户这些内容。这种处理管道基于单个主题创建实时数据流图。从0.10.0.0开始，Apache Kafka提供了一个轻量级但功能强大的流处理库，名为Kafka Streams，用于执行上述的数据处理。除了Kafka Streams，其他开源流处理工具包括Apache Storm和Apache Samza。 溯源是一种应用程序设计风格，其中将状态更改记录为按时间顺序排列的记录序列。Kafka支持非常大的存储日志数据，这使得它成为这种风格的应用程序的后端。 Kafka可以作为分布式系统的一种外部提交日志。日志有助于在之间数据，并充当故障的重新同步机制，以恢复它们的数据。Kafka的日志压缩特性支持这种用法。在这种用法中，Kafka类似于Apache BookKeeper项目。 卡夫卡是指奥匈帝国作家弗兰兹·卡夫卡。 弗兰茨·卡夫卡（FranzKafka，1883年7月3日—1924年6月3日），20世纪德文家。常采用寓言体，背后的寓意人言人殊，其作品很有深意地抒发了他愤世嫉俗的决心和勇气，别开生面的手法，令二十世纪各个写作流派纷纷追认其为先驱。 代表作是：长篇《美国》《审判》《城堡》，短篇《长城的建造》四、kafka起源《判决》《饥饿艺术家》。 卡夫卡-艺术特点 卡夫卡被认Kafka最初的用例是能够重建一个用户活动跟踪管道，作为一组实时发布-提要。这意味着站点活动(页面浏览、搜索或用户可能采取的其他作)被发布到中心主题，每个活动类型有一个主题。这些提要可用于一系列用例，包括实时处理、实时监视和加载到Hadoop或脱机数据仓库系统以进行脱机处理和报告。为是现代派文学的鼻祖，是表现主义文学的先驱，其作品主题曲折晦涩，情节支离破碎，思路不连贯，跳跃性很大，语言的象征意义很强，这给阅读和理解他的作品带来了一定的困难。 卡夫卡的作品难读，连母语是德语的读者也觉得读懂这些作品不是件容易的事，但他那独到的认识，深刻的批判，入木三分的描写，都深深地吸引着我们，只要你能读进去，只要你能摸到作品的脉络，定会获益匪浅。下点功夫读一读卡夫卡是值得的。 卡夫卡笔下描写的都是生活在下层的小人物，他们在这充满矛盾、扭曲变形的世界里惶恐，不安，孤独，迷惘，遭受压迫而不敢反抗，也无力反抗，向往明天又看不到出路。看到他为我们描绘出的一幅幅画卷我们会感到一阵阵震惊和恐惧。 因为他仿佛在为人类的明天敲起阵阵急促的警钟，他为人类的未来担忧。每位读者在读卡夫卡时都会有自己的感触、理解、认识、联想，但我们希望读者不要迷惘在他所描绘的迷惘中。 1.消息持久化：采用时间复杂度O(1)的磁盘存储结构，即使TB级以上数据也能保证常数时间的访问速度。 2.高吞吐：Kafka拥有很高的吞吐量，即使是在单性能比较低下的商用集群中，也能保证单每秒10万条消息的传输。 3.高容错：Kafka在设计上支持多分区、多副本的策略，拥有很强的容错性。 4.易扩展：在进行集群扩展时，集群无需停机，就可以轻松完成对集群规模的扩展，新增加的自动感知，加入集群。 5.多种处理模式：支持离线、实时的处理模式，应对各种生产环境。 Broker ：一个Broker就是Kafka集群中的一个，多个Broker组成了Kafka集群。 Topic ：Topic只是一个逻辑上的概念。Kafka把同一类数据进行汇总，每一类数据的就是一个Topic。生产者Producer将同一类型的数据写入同一个Topic，消费者Consumer从同一个Topic中消费该同类数据。 Replication ：分区的副本，每个副本存储在不同的Broker中。 Producer ：消息生产者，是Kafka中向Broker发布消息的客户端。 Consumer ：消息消费者，是消费Broker中信息的客户端。 Consumer Group ：Kafka中支持将多个消费者作为一个群体，就是Consumer Group消费者组。每一个Consumer都隶属于一个特定的消费者组，并且一条消息可以发送给多个不同的消费者组，但是每一个消费者组中只能有一个消费者消费该消息。 Zookeeper ：Zookeeper负责保存Kafka的元数据，同时也负责Kafka的集kafka_consumer = KafkaConsumer(群管理。 4 Consumer Group消费者组是为了加快消费的读取速度的一个模型，一个消费者组中的多个Consumer可以并行消费同一个Topic中的数据。 并且多个Consumer Group可以消费同一个Topic，这些消费者组之间是平等的，即同一条消息可同时被多个消费者组消费。 同一个Consumer Group消费者组中的多个Consumer消费者之间是竞争关系，也就是说同一条消息在一个消费者组中只能被一个消费者所消费。 Partition是一个物理结构，它的实际存储在一个文件夹目录内，目录中包含若干个Segment文件。 Segment文件是Kafka中的最小存储单元，它是由以Message在Partition中的起始偏移量命名的数据文件（ .log）和索引文件（ .index, .timeindex）组成，如图所示。 早期的Kafka版本里并没有副本的概念，这样一旦某个宕机，那么这台上存储的所有数据都有可能丢失，为了解决这个问题，在之后的版本更新中提出了Replication副本的概念。 之前讲到过，一个Topic主题中可以有多个Partition分区，并且每个分区都可以有多个副本，这是Kafka的一个数据冗余机制。除此之外，Kafka为了保证多个副本的数据一致性，从同一个分区的多个副本中选举出一个Partition Leader，由这个Leader来负责读写，其他的副本作为Follower从Leader中同步消息，通过这样一个副本同步机制，保证了多副本的数据一致性。 同样的Kafka中的Broker也会有这样的一个选举机制。每个Broker在启动时都会创建一个Kafka Controller进程。由这个Kafka Controller以及Zookeeper来选举出一个Kafka Controller Leader。Kafka Controller Leader负责管理Kafka集群的分区和副本状态，避免了分区副本直接在Zookeeper上注册Watcher和竞争创建临时Znode，导致Zookeeper集群的负载过重。 Kafka的命令作以对Topic为主，通过PPT中的例子可以创建一个带有Partition以及制定了Replication的Topic，注意，在创建Topic时要制定Topic的名字以及Kafka对应的以及端口。 在Data Syndrome，我们使用并喜爱Kafka。它使我们能够以最少的努力和复杂性将批处理变为实时处理。然而，在最近的一个项目中，我们学到了有关kafka-python软件包的惨痛教训，该软件包促使我思考该如何选择开源工具。本文将反思我们的开源决策过程，介绍两个用于Python的Kafka客户端、我们遇到的问题及我们将采用的解决方案。 kafka-python：蛮荒的西部 kafka-python是的Kafka Python客户端。我们过去使用时从未出现过任何问题，在我的《敏捷数据科学2.0》一书中我也用过它。然而在最近这个项目中，它却出现了一个的问题。我们发现，当以文档化的方式使用KafkaConsumer、Consumer迭代式地从消息队列中获取消息时，最终到达主题topic的由Consumer携带的消息通常会丢失。我们通过控制台Consumer的分析验证了这一点。 需要更详细说明的是，kafka-python和KafkaConsumer是与一个由SSL保护的Kafka服务（如Aiven Kafka）一同使用的，如下面这样： topic, enable_auto_commit=True, bootstrap_servers=config.kafka.host, api_version=(0, 10), security_protocol='SSL', ssl_check_hostname=True, ssl_cafile=config.kafka.ca_pem, ssl_certfile=config.kafka.serv_cert, ssl_keyfile=config.kafka.serv_key )for message in kafka_consumer: application_message = json.loads(message.value.decode()) ... 当以这样的方式使用时，KafkaConsumer会丢失消息。但有一个变通方案，就是保留所有消息。这个方案是Kafka服务提供商Aiven support提供给我们的。它看起来像这样： while True: raw_messages = consumer.poll(timeout_ms=1000, max_records=5000) for topic_partition, messages in raw_messages.s(): application_message = json.loads(message.value.decode()) ... 虽然这个变通方案可能有用，但README中的方丢弃消息使我对其失去兴趣。所以我找到了一个替代方案。 confluent-kafka：企业支持 发现coufluent-kafka Python模块时，我感到无比惊喜。它既能做librdkafka的外封装，又非常小巧。librdkafka是一个用C语言写的kafka库，它是Go和.NET的基础。更重要的是，它由Confluent公司支持。我爱开源，但是当“由非正式社区拥有或支持”这种方式效果不行的时候，或许该考虑给替代方案印上公章、即该由某个公司拥有或支持了。不过，我们并未购买商业支持。我们知道有人会维护这个库的软件质量，而且可以选择买或不买商业支持，这一点真是太棒了。 用confluent-kafka替换kafka-python非常简单。confluent-kafka使用poll方法，它类似于上面提到的访问kafka-python的变通方案。 kafka_consumer = Consumer( {"api.version.request": True, "enable.automit": True, "group.id": group_id, "security.protocol": "ssl", "ssl.ca.location": config.kafka.ca_pem, "ssl.certificate.location": config.kafka.serv_cert, "ssl.key.location": config.kafka.serv_key, "": {"auto.offset.reset": "allest"} }) consumer.subscribe([topic]) # Now loop on the consumer to read messages running = True while running: message = kafka_consumer.poll() application_message = json.load(message.value.decode()) 现在我们能收到所有消息了。我并不是说kafka-python工具不好，我相信社区会对它的问题做出反应并解决。但从现在开始，我会一直坚持使用confluent-kafka。 开源治理 开源是强大的，但是涉及到复杂的“大数据”和NoSQL工具时，通常需要有一家大公司在背后推动工具的开发。这样你就知道，如果那个公司可以使用工具，那么该工具应该拥有很好的基本功能。它的出现可能是非正式的，就像某公司发布类似FOSS的项目一样，但也可能是正式的，就像某公司为工具提供商业支持一样。当然，从另一个角度来看，如果一家与开源社区作对的公司负责开发某个工具，你便失去了控制权。你的意见可能无关紧要，除非你是付费客户。 理想情况是采取开源治理，就像Apache基金会一样，还有就是增加可用的商业支持选项。这对互联网上大部分的免费软件来说根本不可能。限制自己只使用那些公司盖章批准后的工具将非常限制你的自由。这对于一些商店可能是正确选择，但对于我们不是。我喜欢工具测试，如果工具很小，而且只专心做一件事，我就会使用它。 信任开源 对于更大型的工具，以上决策评估过程更为复杂。通常，我会看一下提交问题和贡献者的数量，以及一次commit的日期。我可能会问朋友某个工具的情况，有时也会在推特上问。当你进行嗅探检查后从Github选择了一个项目，即说明你信任社区可以产出好的工具。对于大多数工具来说，这是没问题的。 但信任社区可能存在问题。对于某个特定的工具，可能并没有充分的理由让你信任社区可以产出好的软件。社区在目标、经验和开源项目的投入时间方面各不相同。选择工具时保持审慎态度十分重要，不要让理想蒙蔽了判断。kafka集群测试正常，但是Ja连接kafka出现异常，急求大神解答！！！！！！！！！！！
Kafka是一种分布式流处理平台和消息队列系统，由Apache软件基金会开发和维护。它最初是由LinkedIn开发的，用于处理大规模的实时数据流。Kafka的设计目标是提供高吞吐量、低延迟、可持久化的消息传递系统，以满足现代数据处理和分析的需求。kafka
- partition的每个副本都必须位于单独的broker上kafka中的topic为什么要进行分区
Partition ：分区是一个物理概念，每一个Topic都可以包含很多个Partition。是一个有序的不可修改的消息队列，每个分区内消息是有序的，并且每个分区对应一个文件夹，用来存储分区的数据以及索引。Kafka使用场景
Kafka-分布式发布-消息系统，最初是由LinkedIn公司所开发，Scala语言编写，之后加入Apache旗下，成为了Apache的子项目。卡夫卡是什么
5-kafka(分布式消息队列)
kafka在设计之初就是为了针对大数据量的传输处理，高吞吐量、低延迟最主要看的就是单位时间内所能读写的数据总量，我们先来看生产端。如何在kafka-python和confluent-kafka之间做出选择
另外一种情况，消费组中的消费者数量多于分区数，如下图所示
声明：本站所有文章资源内容，如无特殊说明或标注，均为采集网络资源。如若本站内容侵犯了原著者的合法权益，可联系 836084111@qq.com 删除。