kafka版本怎么看（《深入浅出》Apache Kafka实战）

本文目录

• 《深入浅出》Apache Kafka实战
• 如何查看kafka命令 找不到
• cdh5.10对应哪个版本的kafka
• Kafka升级
• linux 怎样查看kafka的某 topic数据
• 一文解密Kafka，Kafka源码设计与实现原理剖析，真正的通俗易懂
• kafka-docker上使用+常用指令

《深入浅出》Apache Kafka实战

2011年年初，美国领英公司(LinkedIn) 开源了一款基础架构软件，以奥地利作家弗兰兹.卡夫卡(Franz Kafka)的名字命名，之后LinkedIn 将其贡献给Apache基金会，随后该软件于2012年10月成功完成孵化并顺利晋升为Apache顶级项目---这 便是大名鼎鼎的ApacheKafka。历经7年发展，2017年 11月，Apache Kafka正式演进到1.0时代，本书就是基于1.00版本来展开介绍Kafka的设计原理与实战的。

给大家分享一篇Apache Kafka实战的PDF

第1章：认识Apache Kafka

Kafka快速入门

消息引擎系统

Kafka概要设计

Kafka基本概念与术语

Kafka使用场景.

本章小结

第二章:Kafka发展历史

Kafka的历史

Kafka版本变迁

如何选择Kafka版本

Kafka与Confluent

本章小结

第3章：Kafka线上环境部署

集群环境规划

伪分布式环境安装

多节点环境安装

验证部署

参数设置

本章小结

第4章：producer开发

producer概览

构造producer

消息分区机制

消息序列化

producer拦截器

无消息丢失配置

消息压缩.

多线程处理

旧版本producer

本章小结

第五章：consumer开发

consumer概览

构建consumer

订阅topic

消息轮询.

位移管理

重平衡

解序列化

多线程消费实例

独立consumer

旧版本consumer

本章小结

第六章：Kafka设计原理

broker端设计架构

producer端设计

consumer 端设计

实现精确一次处理语义

本章小结

第七章：管理Kafka集群

集群管理

topic管理

topic动态配置管理

cosumer相关管理

topic分区管理

Kafka常见脚本工具.

..........

常见问题

本章小结

第八章：监控Katka集群

集群健康度检查

MBean监控

broker端JMX监控

clients端JMX监控

JVM监控

Os监控.

主流监控框架

本章小结

第九章：调优Kafka集群

引言

确定调优目标

集群基础调优

调优吞吐量

调优延时

调优持久性

调优可用性

本章小结

第十章：Kafka Connect与Kafka Streams

引言

Kafka Connect

Kafka Streams

本章小结

这篇PDF的内容太多了我没有办法一一展示出来，我已经把这篇 PDF整理好了，需要免费领取的朋友麻烦转发这篇文章，然后私信【学习或者JVM】即可。

如何查看kafka命令 找不到

基于0.8.0版本。
##查看topic分布情况kafka-list-topic.sh
bin/kafka-list-topic.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 （列出所有topic的分区情况）
bin/kafka-list-topic.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --topic test （查看test的分区情况）
其实kafka-list-topic.sh里面就一句
exec $(dirname $0)/kafka-run-class.sh kafka.admin.ListTopicCommand $@
实际是通过
kafka-run-class.sh脚本执行的包kafka.admin下面的类
##创建TOPIC kafka-create-topic.sh
bin/kafka-create-topic.sh --replica 2 --partition 8 --topic test --zookeeper 192.168.197.170:2181,192.168.197.171:2181
创建名为test的topic， 8个分区分别存放数据，数据备份总共2份
bin/kafka-create-topic.sh --replica 1 --partition 1 --topic test2 --zookeeper 192.168.197.170:2181,192.168.197.171:2181
结果 topic: test2 partition: 0 leader: 170 replicas: 170 isr: 170
##重新分配分区kafka-reassign-partitions.sh
这个命令可以分区指定到想要的--broker-list上
bin/kafka-reassign-partitions.sh --topics-to-move-json-file topics-to-move.json --broker-list "171" --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --execute
cat topic-to-move.json
{"topics":
,
"version":1
}
##为Topic增加 partition数目kafka-add-partitions.sh
bin/kafka-add-partitions.sh --topic test --partition 2 --zookeeper 192.168.197.170:2181,192.168.197.171:2181 （为topic test增加2个分区）
##控制台接收消息
bin/kafka-console-consumer.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --from-beginning --topic test
##控制台发送消息
bin/kafka-console-producer.sh --broker-list 192.168.197.170:9092,192.168.197.171: 9092 --topic test
##手动均衡topic, kafka-preferred-replica-election.sh
bin/kafka-preferred-replica-election.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --path-to-json-file preferred-click.json
cat preferred-click.json
{
"partitions":
[
{"topic": "click", "partition": 0},
{"topic": "click", "partition": 1},
{"topic": "click", "partition": 2},
{"topic": "click", "partition": 3},
{"topic": "click", "partition": 4},
{"topic": "click", "partition": 5},
{"topic": "click", "partition": 6},
{"topic": "click", "partition": 7},
{"topic": "play", "partition": 0},
{"topic": "play", "partition": 1},
{"topic": "play", "partition": 2},
{"topic": "play", "partition": 3},
{"topic": "play", "partition": 4},
{"topic": "play", "partition": 5},
{"topic": "play", "partition": 6},
{"topic": "play", "partition": 7}
]
}
##删除topic,慎用，只会删除zookeeper中的元数据，消息文件须手动删除
bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic test666 --zookeeper 192.168.197.170:2181 ,192.168.197.171:2181

cdh5.10对应哪个版本的kafka

kafka没有像别的一样有，kafka -verison的命令，但是你可以查看kafka/libs 下的库文件，知道kafka的版本号：
find ./libs/ -name \*kafka_\* | head -1 | grep -o ’\kafka*’kafka_2.9.2-0.8.1.1.jar.asc

Kafka升级

kafka-0.11.0.x引入了一个新的消息格式版本和线协议，通过遵循以下的滚动升级计划，可以实现不停服务完成升级。

从0.10.2版本开始，Java客户端(生产者和消费者)是可以和旧版本通信的，而0.11.0客户端只可以与0.10.0之后的版本通信。如果现有集群的broker老于0.10.0版本，那么必须在升级客户端之前，先升级kafka集群中所有的broker。0.11.0版本的kafka broker支持0.8.x以上版本的客户端

1. 更新所有broker的server.properties文件，并添加以下属性：

inter.broker.protocol.version=CURRENT_KAFKA_VERSION (如：0.8.2.0, 0.9.0.0或0.10.0.0).

log.message.format.version=CURRENT_KAFKA_VERSION (有关此配置的详细信息，请查看升级后潜在的性能影响。)

2. 每次升级一个broker：关闭broker，替换新版本，然后重新启动它。

3. 一旦整个群集升级，通过server.properties中的inter.broker.protocol.version并将其设置为0.11.0.x来转换所有协议，但现在不要修改log.message.format.version

4. 逐个重新启动broker，使新协议版本生效。

5. 当所有的消费者升级到0.11.0以上版本时，再修改kafka broker的server.properties的log.message.format.version为0.11.0.然后再逐个重启broker。注意旧版本的消费者是不支持新版本消息格式的，所以为了避免下变频的性能开销，尽量使用新的Java消费者。

1.如果可接受停机，你可以简单地将所有broker关闭，更新版本并重启启动，它们将默认从新版本开始。

2.变换协议版本和重启启动可以在broker升级完成后的任何时间去做，不必马上做。

3.在变更log.message.format.version全局设置之前依旧可以使用topic管理工具(bin/kafka-topics.sh)

4.如果是从0.10.0之前的版本升级，那么massage format不必要在0.11.0之前先更新设置0.10.0

linux 怎样查看kafka的某 topic数据

1、创建一个需要增加备份因子的topic列表的文件，文件格式是json格式的。

2、使用kafka官方提供的工具拿到上面topic的partions 分布情况，并重定向到文件中。

3、修改ressgintopic.conf 文件的，手动分配新增加的partion 备份因子。

4、通过下面命令执行备份因子扩容过程，bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json。

5、最后查看kafka的某 topic数据如图。

注意事项：

Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

一文解密Kafka，Kafka源码设计与实现原理剖析，真正的通俗易懂

Apache Kafka （简称Kafka ）最早是由Linkedln开源出来的分布式消息系统，现在是Apache旗下的一个子项目，并且已经成为开册、领域应用最广泛的消息系统之 Kafka社区也非常活跃，从 版本开始， Kafka 的标语已经从“一个高吞吐量、分布式的消息系统”改为“一个分布式的流平台”
关于Kafka，我打算从入门开始讲起，一直到它的底层实现逻辑个原理以及源码，建议大家花点耐心，从头开始看，相信会对你有所收获。

作为 个流式数据平台，最重要的是要具备下面 个特点

消息系统：
消息系统 也叫作消息队列）主要有两种消息模型：队列和发布订Kafka使用消费组（ consumer group ）统 上面两种消息模型 Kafka使用队列模型时，它可以将处理 作为平均分配给消费组中的消费者成员

下面我们会从 个角度分析Kafka 的几个基本概念，并尝试解决下面 个问题

消息由生产者发布到 fk 集群后，会被消费者消费 消息的消费模型有两种：推送模型（ pu和拉取模型（ pull 基于推送模型的消息系统，由消息代理记录消费者的消费状态 消息代理在将消息推送到消费者后 标记这条消息为已消费

但这种方式无法很好地保证消息的处理语义 比如，消息代理把消息发送出去后，当消费进程挂掉或者由于网络原因没有收到这条消息时，就有可能造成消息丢失（因为消息代理已经 这条消息标记为自己消费了，但实际上这条消息并没有被实际处理） 如果要保证消息的处理语义，消息代理发送完消息后，要设置状态为“已发送”，只有收到消费者的确认请求后才更新为“已消费”，这就需要在消息代理中记录所有消息的消费状态，这种做法也是不可取的

Kafka每个主题的多个分区日志分布式地存储在Kafka集群上，同时为了故障容错，每个分区都会以副本的方式复制到多个消息代理节点上 其中一个节点会作为主副本（ Leader ），其 节点作为备份副本（ Follower ，也叫作从副本）

主副本会负责所有的客户端读写操作，备份副本仅仅从主副本同步数据 当主副本 IH 现在故障时，备份副本中的 副本会被选择为新的主副本 因为每个分区的副本中只有主副本接受读写，所以每个服务端都会作为某些分区的主副本，以及另外一些分区的备份副本这样Kafka集群的所有服务端整体上对客户端是负载均衡的

消息系统通常由生产者「pro ucer 消费者（ co sumer ）和消息代理（ broke 大部分组成，生产者会将消息写入消息代理，消费者会从消息代理中读取消息 对于消息代理而言，生产者和消费者都属于客户端：生产者和消费者会发送客户端请求给服务端，服务端的处理分别是存储消息和获取消息，最后服务端返回响应结果给客户端

新的生产者应用程序使用 af aP oduce 对象代表 个生产者客户端进程 生产者要发送消息，并不是直接发送给 务端 ，而是先在客户端 消息放入队列 然后 一个 息发送线程从队列中消息，以 盐的方式发送消息给服务端 Kafka的记 集器（ Reco dACCUl’lUlato ）负责缓存生产者客户端产生的消息，发送线程（ Sende ）负责读取 集器的批 过网络发送给服务端为了保证客户端 络请求 快速 应， Kafka 用选择器（ Selecto 络连接 读写 理，使网络连接（ Netwo kCl i.ent ）处理客户端 络请求

追加消息到记录收集器时按照分区进行分组，并放到batches集合中，每个分区的队列都保存了将发送到这个分区对应节点上的 记录，客户端的发送线程可 只使用 Sende 线程迭 batches的每个分区，获取分区对应的主剧本节点，取出分区对应的 列中的批记录就可以发送消息了

消息发送线程有两种消息发送方式 按照分区直接发送 按照分区的目标节点发迭 假设有两台服务器， 题有 个分区，那么每台服务器就有 个分区 ，消息发送线程迭代batches的每个分 接往分区的主副本节点发送消息，总共会有 个请求 所示，我 先按照分区的主副本节点进行分组， 属于同 个节点的所有分区放在一起，总共只有两个请求做法可以大大减少网络的开销

消息系统由生产者 存储系统和消费者组成 章分析了生产者发送消息给服务端的过程，本章分析消费者从服务端存储系统读取生产者写入消息的过程 首先我 来了解消费者的 些基础知识

作为分布式的消息系统， Kafka支持多个生产者和多个消费者，生产者可以将消息发布到集群中不同节点的不同分区上；「肖费者也可以消费集群中多个节点的多个分区上的消息 写消息时，多个生产者可以 到同 个分区 读消息时，如果多个消费者同时读取 个分区，为了保证将日志文件的不同数据分配给不同的消费者，需要采用加锁 同步等方式，在分区级别的日志文件上做些控制

相反，如果约定“同 个分区只可被 个消费者处理”，就不需要加锁同步了，从而可提升消费者的处理能力 而且这也并不违反消息的处理语义：原先需要多个消费者处理，现在交给一个消费者处理也是可以的 3- 给出了 种最简单的消息系统部署模式，生产者的数据源多种多样，它们都统写人Kafka集群 处理消息时有多个消费者分担任务 ，这些消费者的处理逻辑都相同， 每个消费者处理的分区都不会重复

因为分区要被重新分配，分区的所有者都会发生变 ，所以在还没有重新分配分区之前 所有消费者都要停止已有的拉取钱程 同时，分区分配给消费者都会在ZK中记录所有者信息，所以也要先删ZK上的节点数据 只有和分区相关的 所有者 拉取线程都释放了，才可以开始分配分区

如果说在重新分配分区前没有释放这些信息，再平衡后就可能造成同 个分区被多个消费者所有的情况 比如分区Pl 原先归消费者 所有，如果没有释放拉取钱程和ZK节点，再平衡后分区Pl 被分配给消费者 了，这样消费者 和消费者 就共享了分区Pl ，而这显然不符合 fka 中关于“一个分区只能被分配给 个消费者”的限制条件 执行再平衡操作的步骤如下

如果是协调者节点发生故障，服务端会有自己的故障容错机制，选出管理消费组所有消费者的新协调者节，点消费者客户端没有权利做这个工作，它能做的只是等待一段时间，查询服务端是否已经选出了新的协调节点如果消费者查到现在已经有管理协调者的协调节点，就会连接这个新协调节，哉由于这个协调节点是服务端新选出来的，所以每个消费者都应该重新连接协调节点

消费者重新加入消费组，在分配到分区的前后，都会对消费者的拉取工作产生影响 消费者发送“加入组请求”之前要停止拉取消息，在收到“加入组响应”中的分区之后要重新开始拉取消息时，为了能够让客户端应用程序感知消费者管理的分区发生变化，在加入组前后，客户端还可以设置自定义的“消费者再平衡监听器”，以便对分区的变化做出合适的处理

kafka-docker上使用+常用指令

生产者向broker发送消息，消费者接收消息，broker是物理概念，部署几个kafka即几个broker，topic是逻辑概念，往topic里发送消息会发送到设置好的几个partion上,每个partion存储作为不同队列存储不同数据，partion有leader和follower备份机制，消息发送时会轮循发送到不同broker的不同partion中，同一消费者只能消费同一分区，通过offset记录消费位置，消费者组可以访问一个topic的不同partion

启动镜像

启动kafka可以带上参数，这样会自动修改kafka里的配置文件（/opt/kafka_版本/conf/server.properties）,否则不带参数需要自己进入进行手动修改 带参数版启动可参考

其中172.17.0.3需要改成自己docker的网桥连接地址

查看已启动容器

查看所有容器

启动未启动的容器

进入kafka容器

创建主题

主题和分区可以理解为：topic是逻辑划分，kafka通过topic进行区分消息，topic的数据会被存储到日志中，如果数据量太大可以引入partion（同时提高读写吞吐量）来分段存储数据。其中replication-factor作用是将任意分区复制到broker上，broker是物理概念，部署了一个kafka可认为broker数为1，我本机只有一个kafka所以这里replication-factor超过1会报错。 综上几个概念可以理解为：集群中有多个broker，创建主题时可以指明topic有多个partitions(消息拆分到不同分区进行存储，一个partion只能被一个消费者消费--partion内部保证接收数据顺序)，可以为分区创建多个副本replication，不同副本在不同的broker中（作为备份使用，这里有leader和flower的区分） 。

查看topic信息

集群部署
可以通过compose集群化部署过es，这里通过创建另一个compose.yml文件来部署kafka，配置文件参考 docker-compose集群部署

生产者：

消费者：
方式一：从当前主题的迁移量位置+1开始取数据

方式二：从当前主题第一条消息开始消费

生产者将消息发送broker，broker将消息保存到本地日志中，消息的保存时有序的
单播消息：
当存在一个生产者，一个消费者组的时候，一个消费者组中只有一个消费者会收到消息

多播消息：
当存在一个生产者，多个消费组，不同消费组只有一个消费者收到消息

查看消费组详细信息：

CURRENT-OFFSET:最后被消费的偏移量
LOG-END-OFFSET：消息总量（最后一条消息的偏移量）
LAG ：积压了多少条消息

常见问题：
1、如何防止消息丢失
生产者：使用同步消息发送；ack设置为1/all;设置同步分区数》=2
消费者：把自动提交改成手动提交
2、如何防止消息的重复消费
针对网络抖动导致的生产者重试（发送消息），可以设置消费者加锁解决；
3、消息积压
消费者使用多线程异步处理接收数据；创建多个消费者组部署到其他机器上；通过业务架构设计，提升业务层面消费性能。

ps:
缓冲区：kafka默认会创建一个消息缓冲区去存放要发送的消息，大小是32M，每次本地线程会去缓冲区拉16K数据发送到broker，如果不到16K等待10ms也会将数据发送到broker

参考链接：
1、kafka安装教程--推荐
2、kafka配置文件server.properties参数说明
3、创建主题分区数
4、解决docker容器启动不了的问题
5、通过docker-compose集群部署
6、学习视频