Kafka 应用入门

Kafka 应用入门

Kafka 是 linkedin 使用 Scala 编写具有高水平扩展和高吞吐量的分布式消息系统。

一：Kafka 重要概念：

• 对消息保存时根据 Topic 进行归类

• 发送消息者成为 Producer

• 消息接受者成为 Consumer

• kafka 集群有多个 kafka 实例组成，每个实例(server)称为 broker

无论是 Kafka集群，还是 producer 和 consumer 集群都依赖于 zookeeper 来保证系统可用性，为集群保存一些 meta 信息。

二：Kafka的定位

• 消息中间件
• 消息引擎
• 分布式实时流处理平台

使用场景：

• 大数据领域：网站行为分析、日志聚合、应用监控、流式数据处理、在线和离线数据分析等领域。
• 数据集成：将消息导入 OSS、RDS、Hadoop、HBase等离线数据仓库。
• 流计算集成：与Spark、Storm等流计算引擎集成。

三、AMQP协议相关概念

AMQP (Advanced Message Queuing Protocol) ，是一个提供统一消息服务的标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件而设计

基本的概念：

• AMQP服务器端（broker）：用来接收生产者发送的消息并将这些消息路由给服务器中的队列
• 消费者（Consumer）：从消息队列中请求消息的客户端应用程序
• 生产者（Producer）：向 broker 发布消息的客户端应用程序
  

四：Kafka 与ZK的关系

在安装Kafka的时候，必须依赖ZK的服务，在生产环境通常是ZK的集群，而且Kafka自带了ZK。
ZK中的储存（通过可视化工具ZooViewer）

总结：利用zk的有序节点、临时节点、监听制作，帮kafka管理broker、topic、partition、consumer的信息，包括元数据的变动，负载均衡、命名服务、分布式通知、集群管理、选举、分布式锁等

五：kafka架构分析

5.1、Broker

kafka 作为一个中间件，是存储和转发消息的，有点像中介，所以每个kafka实例(server)称为 broker，默认端口9092，生产者消费者都需要和broker建立连接。

5.2、消息

传输的数据叫做消息，或者叫做记录(Record)，在客户端代码中Record可以是KV的键值对。
消息在传输过程中需要序列化，需要指定序列化工具。
消息在服务端存储格式有两种 RecordBatch和 Record。

5.3、生产者

顾名思义是发生消息的一方，生产者往某个Topic上发布消息，生产者也负责选择发布到Topic上的哪一个分区。最简单的方式从分区列表中轮流选择，也可以根据某种算法依照权重选择分区。开发者负责如何选择分区的算法。

为了提升发送的效率，生产者是批量发送，发送是由batchsize 和 lingerms 参数决定。

5.4、消费者

消费者获取消息由两种模式：

• pull：消息放在broker，消费者自己决定什么时候去拉取
• push：消息放在consumer，主要有消息到达broker，直接推送给消费者

kafka 只有pull模式，因为在push模式下，如果消息生产的速度大于消费的速率，那么消费者肯定处理不完，直达挂掉。pull模式 下消费者可以自己控制一次获取多少条数据：max.poll.records

消费者使用一个消费组名称来进行标识，发布到 topic 中的每条记录被分配给订阅消费组中的一个消费者实例。消费者实例可以分布在多个进程中或者多个机器上。

• 如果所有的消费者实例在同一消费组中，消息记录会负载平衡到每一个消费者实例。
• 如果所有的消费者实例在不同的消费组中，每条消息记录会广播到所有的消费者进程
  

5.5、Topics 和 Logs

Topic 就是数据主题，是数据记录发布的地方，可以用来区分业务系统。Kafka 中的 Topics 总是多订阅者模式，一个 topic 可以拥有一个或者多个消费者来订阅它的数据。

对于每一个topic， Kafka集群都会维持一个分区日志

5.5、Partition

如果Topic中的数据太多了，两个问题：

• 不方便横向扩展：比如想把数据分布在不同的集群节点
• 负载：所有客户端都 *** 作同一个topic，高并发的场景下性能大大下降

所以，把topic进行拆分（分片），这就是Partition分区，一个topic可以分多个Partition，在创建topic时指定，如果没有指定分区数，默认的分区数一个 num.partitions=1

5.6、Replication

如果Partition的数据只存储一份，在发生网络问题或者硬件故障的时候，该分区的数据就无法访问，所以在0.8版本之后增加了副本机制，这是一种冗余备份策略。

每个Partition可以有若干个副本，由replication-factor 指定一个topic的副本数。

• 同一个partition的多个replication不允许在同一broker上
• 每个partition的replication中，有一个leader ，零或多个follower
• leader处理此分区的所有的读写请求， follower仅仅被动的复制数据
• leader宕机后，会从follower中选举出新的leader

总结：

3台Broker。两个Topic：Topic0和Topic1。
Topic0有2个分区：partition0和partition1，每 个分区一共3个副本。
Topic1只有1个分区：partition0，每个分区一共3 个副本。
图中红色字体的副本代表是leader，黑色字体的副 本代表是follower。
绿色的线代表是数据同步。蓝色的线是写消息，橙 色的线是读消息，都是针对leader节点。

有两个消费者组，
第一个消费者组，消费了topic0 的两个分区。
第二个消费者组，既消费topic0，又消费topic1。
其中有一个消费者，消费topic0的partition0，还 消费topic1的partition0。
有一个消费者，消费 partition0的partition1。有一个消费者，没有 partition可以消费。

六：Kafka Java开发

四个核心 API：

• Producer API
  • 允许一个应用程序发布一串流式的数据到一个或者多个 Kafka topic。
• Consumer API
  • 允许一个应用程序订阅一个或多个 topic ，并且对发布给他们的流式数据进行处理。
• Streams API
  • 允许一个应用程序作为一个流处理器，消费一个或者多个 topic 产生的输入流，然后生产一个输出流到一个或多个 topic中去，在输入输出流中进行有效的转换。
• Connector API
  允许构建并运行可重用的生产者或者消费者，将Kafka topics连接到已存在的应用程序或者数据系统。比如，连接到一个关系型数据库，捕捉表（table）的所有变更内容。

6.1、Producer API

Properties props = new Properties();
props.put("batch.size",16384);   //默认值为16384
props.put("linger.ms",16384);   //默认值为0
props.put("acks", "all");
props.put("retries",1);
//... 
Producer<String, String> producer = new KafkaProducer(props);
ProducerRecord<String, String> record =
  new ProducerRecord<String, String>("my-topic", "key", "value");
  producer.send(record);
  producer.close();

• Producer会为每个partition维护一个缓冲，用来记录还没有发送的数据，每个缓冲区大小用 batch.size指定，默认值为16k
• linger.ms为，buffer中的数据在达到batch.size前，需要等待的时间
• acks用来配置请求成功的标准

6.2、Consumer API

@KafkaListener

• Simple Cnsumer
  位于kafka.javaapi.consumer包中，不提供负载均衡、容错的特性每次获取数据都要指定topic、partition、offset、fetchSize
• High-level Consumer
  该客户端透明地处理kafka broker异常，透明地切换consumer的partition，通过和broker交互来实现consumer group级别的负载均衡

七：生产者事务

通过事务，kafka可以保证跨生产者会话的消息幂等发送。

7.1、为什么需要事务

• 发送多条消息，需要这些消息全部成功或者全部失败
• 发送消息到多个topic或者多个partition，需要这些消息全部成功或者全部失败
• 消费以后发出消息 consume-process-produce，从上游接收消息，经过处理后发给下游，这个要保证消息接收和发送同时成功。

生产者和事务相关的方法：

• initTransactions()
• beginTransaction()
• commitTransaction()
• abortTransaction()
• sendOffsetsToTransaction()

事务原理：
1、kafka选择了最常用的2PC
2、既然2pc，必须要有一个协调者角色，Transaction Coordinator
3、事务管理必须要有事务日志，来记录事务的状态 topic__transaction_state
4、如果生产者挂了，事务要在重启后可以继续处理，接着之前未处理完的事务，必须要有一个唯一的生产者事务ID：transaction.id

八：Kafka 与RabbitMQ 对比

• 产品侧重 ，kafka：流式消息处理 ，RabbitMQ ：消息代理
• 性能 ，kafka：更高的吞吐量 ，RabbitMQ ：消息代理
• 消息路由， RabbitMQ 更灵活
• 延迟消息、 死信队列，RabbitMQ 支持 延迟、死信队列
• 消息留存，kafka：消费完之后消息会保留可以设置retention清理消息，
  RabbitMQ ：消费完就删除