啰里吧嗦kafka

1.kafka是什么

kafka官网: http://kafka.apache.org/

kafka是一种高吞吐量的分布式发布订阅消息系统，用它可以在不同系统中间传递分发消息

2.zookeeper是什么

zookeeper是快速、高可用、容错、分布式的协调服务,kafka使用zookeeper用于管理和协调代理,每个kafka代理通过zookeeper协调其他kafka代理

3.kafka和 zookeeper的下载和配置

下载地址：

http://kafka.apache.org/downloads.html

点击Scala 2.11  - kafka_2.11-2.1.1.tgz (asc, sha512) (带src是源代码)

然后点击http://mirrors.shu.edu.cn/apache/kafka/2.1.1/kafka_2.11-2.1.1.tgz

下载完成后解压,放到一个合适的目录下,我放在E:\Kafka\kafka_2.11-2.1.1

1.找到config目录下的server.properties,

设置log.dirs地址,我设置成log.dirs=E:\Kafka\kafka_2.11-2.1.1\kafka-logs

zookeeper.connect=localhost:2181是默认的, 端口和你zookeeper设置的端口保持一致就行

http://zookeeper.apache.org/releases.html#download

点击Active releases may be downloaded from Apache mirrors:Download

进入https://www.apache.org/dyn/closer.cgi/zookeeper/后点击http://mirror.bit.edu.cn/apache/zookeeper/

下载完成后解压,我放在E:\Kafka\zookeeper-3.4.13

1.一般配置文件都在conf目录下,找到zookeeper的conf目录下的zoo_sample.cfg

将其重新命名为zoo.cfg,

修改dataDir地址为自己合适的dataDir=E:\\Kafka\\zookeeper-3.4.13\\tmp

2.类似配置java环境变量一样,设置ZOOKEEPER_HOME, 然后在path中配置%ZOOKEEPER_HOME%\bin

这样在任意目录下通过cmd命令：zkServer, windows系统就能找到对应目录下bin目录下的zkServer.cmd命令并执行

4.windows环境下的kafka配置和使用

在kafka的解压目录下,例如E:\Kafka\kafka_2.11-2.1.1

按住shift在当前目录下进入cmd命令窗口

启动kafka, 命令窗口别关
.\bin\windows\kafka-server-start.bat .\config\server.properties

另开一个命令窗口, 创建一个名为test的topic
.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

查看主题
.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

创建生产者
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test
执行完后可随便输入一个字符串 

创建消费者
.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
执行完命令后会发现 命令行出现了刚才生产的字符串

查看topic详情
.\bin\windows\kafka-topics.bat --describe --zookeeper 127.0.0.1:2181 --topic test



在生产者消费者模式下, 按ctrl+c 退出

启动kafka时出现各种问题和解决, 第一个出现的是错误: 找不到或无法加载主类

这是由于我的java环境由jre换成了jdk,找到kafka_2.12-1.0.0\bin\windows\kafka-run-class.bat

文件,将set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp "%CLASSPATH%" %KAFKA_OPTS% %

%CLASSPATH%前后加上双引号

第二个报错是java version52.0, 通过cmd , java -version命令发现安装版本是1.8, 但是环境变量配置的路径jdk是1.7的,版本不一致, 修改环境变量java_home路径解决

5.kafka节点之间如何复制备份的

备份机制是干啥的： 备份机制保证了kafka集群中的节点挂掉后而不影响整个集群的工作

生产者向topic中发送数据，消费者消费该topic对应的数据，为了提高吞吐量，生产者会将该topic对应的数据分别发送到多个partition，每个partition都有一定数量的副本作为备份，以提高kafka的高可用性

  p0-leader副本   ------- p0-follower副本 
  |
   ----------------------------------------------P0-follower副本

生产者和消费者都只在leader副本上写读数据，三个leader副本平均分配在三个broker上，其他follower副本都只做备份，以防leader宕机，follower副本升级成为leader副本

三个broker之间是有一定的策略进行数据的读写的，follower副本会隔指定的时间去leader副本上读取最新消息，包括元数据和日志消息

所以kafka节点复制备份其实就是复制分区里的leader副本,当生产者发布消息到topic的某个分区时，消息首先被传递到leader副本，然后leader通知follower有新消息过来，follower去leader中拉取消息，一旦有足够的副本收到消息，leader就会提交这个消息，消费者就能消费到这个消息了。

leader负责维护和跟踪同步副本列表中所有follower滞后状态，消息提交之后才被成功复制到所有的同步副本，消息复制延迟受最慢的follower限制，

5.1 follower副本发生故障

如果某个follower落后太多或宕机，leader会把他从isr中剔除出去。那么该副本对应的分区也就称之为同步失效分区，即under-replicated分区,follower重启后会去leader上恢复最新的HW并将日志截断到HW，并继续从leader中获取HW以后的消息，一旦完全赶上leader，副本将被重新加入到ISR队列中，系统将重新回到fully replicated(全量同步)模式。

5.2 leader副本发生故障

leader发生故障，其他follower会争相竞争做leader，最终只有一个follower竞争成功升级成为leader，故障leader重启后成为follower去新leader同步消息 (使用Zookeeper实现leader选举。如果leader失败，controller会从ISR选出一个新的leader

)

注 :broker概念

已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理(Broker). 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

ISR:in-sync replicas

kafka维护的一个副本维护队列，ISR的副本保持和leader的同步，当然leader本身也在ISR中。初始状态所有的副本都处于ISR中，当一个消息发送给leader的时候，leader会等待ISR中所有的副本告诉它已经接收了这个消息，如果一个副本失败了，那么它会被移除ISR。下一条消息来的时候，leader就会将消息发送给当前的ISR中节点了

HW: high watermark

是指ISR中所有节点都已经复制完的消息的offset。也是消费者所能获取到的消息的最大offset

LEO：LogEndOffset，表示每个分区log的最后一条消息的offset

6. kafka消息是否会丢失？为什么？

消息是否会丢失从两个角度来看

6.1消息发送

kafka消息的发送方式分同步(sync)、异步(async)两种方式

生产者如果异步发送，会造成消息丢失，发送的过程中kafka会先把消息缓存起来。然后批量发送。 若批量发送之前client宕机会造成消息丢失。生产者不丢失消息需要同步发送

kafka服务器默认异步刷盘，先刷到系统页缓存，然后再刷新到日志文件。页缓存的数据可能会丢失。解决可以同步的方式刷盘，但是这样效率很低，比rabbitmq低。

配置ack=all ， min.insync.replas > 1 是可以保证页缓存数据不丢失

关闭自动提交？

unclean.leader.election.enable 默认是false 可靠性优先, 不在ISR里的follower不能够参与选举，此时无法进行新的选举，此时整个分区处于不可用状态

6.2消息消费

使用高级接口High-level API，可能存在一个问题就是当消息消费者从集群中把消息取出来、并提交了新的消息offset值后，还没来得及消费就挂掉了，那么下次再消费时之前没消费成功的消息就消失了

简单来说,

producer.type属性进行配置同步异步

request.required.acks属性来确认消息的生产,-1---表示Leader和Follower都接收成功时确认；

同步模式下，确认机制设置为-1，即让消息写入Leader和Follower之后再确认消息发送成功

7. kafka最合理的配置是什么？

8.kafka的leader选举机制是什么？

Kafka将每个Topic进行分区Patition，以提高消息的并行处理，同时为保证高可用性，每个分区都有一定数量的副本 Replica，这样当部分服务器不可用时副本所在服务器就可以接替上来，保证系统可用性。在Leader上负责读写，Follower负责数据的同步。当一个Leader发生故障如何从Follower中选择新Leader呢？

Kafka在Zookeeper上针对每个Topic都维护了一个ISR（in-sync replica---已同步的副本）的集合，集合的增减Kafka都会更新该记录。如果某分区的Leader不可用，Kafka就从ISR集合中选择一个副本作为新的Leader。这样就可以容忍的失败数比较高，假如某Topic有N+1个副本，则可以容忍N个服务器不可用。

如果ISR中副本都不可用，有两种处理方法：

（1）等待ISR集合中副本复活后选择一个可用的副本；

（2）选择集群中其他可用副本；

9. kafka对硬件的配置有什么要求？

磁盘吞吐量 磁盘容量 内存 网络 CPU

10. kafka的消息保证有几种方式？

At most once—Messages may be lost but are never redelivered. 最多一次 --- 消息可能丢失，但绝不会重发。 At least once—Messages are never lost but may be redelivered. 至少一次 --- 消息绝不会丢失，但有可能重新发送。 Exactly once—this is what people actually want, each message is delivered once and only once. 正好一次 --- 这是人们真正想要的，每个消息传递一次且仅一次。