Kafka的介绍和使用

╰+攻爆jí腚メ 2023-02-21 06:46 11阅读 0赞

一：介绍

参考：[kafka][]

kafka是分布式消息系统，以从“一个高吞吐量，分布式的消息系统”改为“一个分布式流平台”。

kafka与传统消息系统不同在于：

*  kafka是一个分布式系统，易于向外扩展；

*  它同时为发布和订阅提供高吞吐量；

*  它支持多订阅者，当失败时能自动平衡消费者；

*  消息的持久化；

二：入门实例

producer：

代码块

Java

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.ProducerRecord;

public class UserKafkaProducer extends Thread{

private final KafkaProducer<Integer,String> producer;

private final String topic;

private final Properties props = new Properties();

public UserKafkaProducer(String topic){

props.put("metadata.broker.list","localhost:9092");

props.put("bootstrap.servers","master2:6667");

props.put("retries",0);

props.put("batch.size",16384);

props.put("linger.ms",1);

props.put("buffer.memory",33554432);

props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

procuder = new KafkaProducer<Integer,String>(props);

this.topic = topic;

}

@Override

public void run(){

int messageNo=1;

while(true){

String messageStr = new String("Message_"+messageNo);

System.out.println("Send:"+messageStr);

//返回的是Future<RecordMetadata>,异步发送

producer.send(new ProducerRecord<Integer,String>(topic,messageStr));

messageNo++;

try{

sleep(3000);

}catch(InterruptedException e){

e.printStackTrace();

}

消费者consumer:

代码块

Java

Properties props = new Properties();

//定义kafka服务的地址，不需要将所有broker指定上

props.put("bootstrap.servers","localhost:9092");

//指定consumer group

props.put("group.id","test");

//是否自动确认offset

props.put("enable.auto.commit","true");

//自动确认offset的时间间隔

props.put("auto.commit.interval.ms","1000");

props.put("session.timeout.ms","30000");

//key的序列化类

props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

//value的序列化类

props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

//定义consumer

KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);

//消费者订阅的topic，可同时订阅多个

consumer.subscribe(Arrays.asList("foo","bar"));

//读取数据，读取超时时间为100ms

while(true){

ConsumerRecords<String,String> records = consumer.poll(100);

for(ConsumerRecord<String,String> record:records){

System.out.println("offset=%d,key=%s,value=%s",record.offset(),record.key(),record.value());

}

三：Kafka架构原理

*  问题：

1）Kafka的topic和分区内部是如何存储的，有什么特点？

2）与传统的消息系统相比，Kafka的消费模型有什么优点？

3）Kafka如何实现分布式的数据存储与数据读取？

*  架构图：

Producer---push-》kafka broker《-pull---consumer

*  Kafka名词解释：

broker：消息中间件处理节点，一个kafka节点就是一个broker，一个或者多个broker可以组成一个kafka集群；

topic：主题，kafka根据topic对消息进行归类，发布到kafka集群的每条消息都需要指定一个topic；

producer：消息生产者，向broker发送消息的客户端；

consumer：消息消费者，从broker读取消息的客户端；

consumer group：每个consumer属于一个特定的consumer group，一条消息可以发送到多个不同的consumer group，但是一个consumer group中只能有一个consumer能够消费该消息；

partition：物理上的概念，一个topic可以分为多个partition，每个partition内部都是有序的；

*  topic和partition：

在kafka中的每条消息都有一个topic，一般不同类型的数据都可以设置不同的主题。一个主题一般会有多个消息的订阅者，当生产者发布消息到某个主题时，订阅了这个主题的消费者都可以接收到生产者写入的新消息。

kafka为每个主题维护了分布式的分区（partition）日志文件，每个partition在kafka存储层面是append log，任何发布到此partition的消息都会被追加到log文件的尾部，在分区中的每条消息都会按照时间顺序分配到一个单调递增的顺序编号，也就是我们的offset，通过offset编号可以确定一条在该partition下的唯一消息。在partition下面保证了有序性，但是topic下面没有保证有序性。

1）如果没有key值，则进行轮询发送；

2）如果有key值，对key值进行hash，然后对分区数量取余，保证了同一个key值的会被路由到同一个分区，如果想队列的强顺序一致性，可以让所有消息设置为同一个key。

*  消费模型

由于推的模型，发送给消费这后，可能由于消费进程挂掉或者网络原因没有收到这条消息，如果标记已发送，则消息会永久消失；如果利用生产者收到消息后回复方法，需要记录消息消费状态，不可取。

kafka采取拉取模型poll，由自己控制消费速度，以及消费进度，消费者可以按照任意的偏移量进行消费。比如消费者可以消费已经消费过的消息今习惯重新处理，或者消费最近的消息等等。

四：producer发布消息

参考学习链接：[kafka][kafka 1]

1）写入方式

producer采用push模式将消息发布到broker，每条消息都被append到partition中，属于顺序写磁盘。

2）消息路由

producer发送消息到broker时，会根据分区算法选择将其存储到哪一个partition。其路由机制为：

指定了partition，则直接使用；

未指定partition但指定key，通过key的value进行hash选择一个partition；

partition和key都未指定，使用轮询选择一个partition；

3）写入流程

1⃣️producer先从zookeeper的/brokers/.../state节点找到该partition的leader；

2⃣️producer将消息发送给该leader；

3⃣️leader将消息写入本地log；

4⃣️followers从leader pull消息，写入本地log后leader发送ACK；

5⃣️leader收到所有ISR中的replica的ACK后，增加HW（high watermark，最后commit的offset），并向producer发送ACK；

4）producer delivery guarantee

参考三种模式代码：[https://blog.csdn.net/laojiaqi/article/details/79034798][https_blog.csdn.net_laojiaqi_article_details_79034798]

1⃣️At most once消息可能会丢，但绝不会重复传输；

2⃣️At least one消息绝不会丢，但可能会重复传输；

3⃣️Exactly once每条消息肯定会被传输依次且仅传输依次；

当producer向broker发送消息时，一旦这条消息被commit，由于replication的存在，它就不会丢。但是如果producer发送数据给broker后，遇到网络问题而造成通信中断，那producer就无法判断该消息是否被commit；虽然kafka无法确定网络故障期间发生了什么，但是producer可以生成一种类似于主键的东西，发生故障时幂等性的重试多次，这样就做到了Exactly once，但是目前没有实现。所以目前默认一条消息从producer到broker是确保了At least once，可以通过设置producer异步发送实现At most once。

五：broker保存消息

1）存储方式

物理上把topic分成一个或多个partition，每个partition物理上对应一个文件，该文件存储该partition的所有消息和索引文件；

2）存储策略

无论消息是否被消费，kafka都会保留所有消息，有两种策略可以删除旧数据：

1⃣️基于时间

2⃣️基于大小

需要注意的是，因为kafka读取特定消息的时间复杂度为O（1），即与文件大小无关，所以这里删除过期文件与提高kafka性能无关；

3）topic创建与删除

topic创建：

1⃣️controller在zookeeper的/brokers/topic节点上注册watcher，当topic被创建，则controller会通过watch得到该topic的partition/replica分配。

2⃣️controller从/brokers/ids读取当前所有可用的broker列表，对于set\_p中的每一个partition：

1）从分配给该partition的所有replica（成为AR）中任选一个可用的broker作为新的leader，并将AR设置为新的ISR；

2）将新的leader和ISR写入/brokers/topics/\[topic\]/partitions/\[partition\]/state;

3⃣️controller通过RPC向相关的broker发送LeaderAndISRRequest。

删除topic：

1⃣️controller在zookeeper的/brokers/topics节点上注册watcher，当topic被删除，则controller会通过watch得到该topic的partition/replica分配。

2⃣️若delete.topic.enable=false,结束；否则controller注册在/admin/delete\_topics上的watch被fire，controller通过回调向对应的broker发送StopReplicaRequest。

六：consumer消费消息

1：kafka提供两套consumer API：

1）The high-level Consumer API;提供从kafka消费数据的高层抽象；

2）The SimpleConsumer API；需要开发人员更多的关注细节；

1.1：The high-level Consumer API提供了consumer group的语义，一个消息只能被group内的一个consumer所消费，且consumer消费消息时不关注offset，最后一个offset由zookeeper保存。

注意：

1）如果消费线程大于partition数量，则有些线程将收不到消息；

2）如果partition数量大于线程数，则有些线程多收到多个partition消息；

3）如果一个线程消费多个partition，则无法保证收到消息的顺序，而一个partition内的消息是有序的。

1.2:The SimpleConsumer API可以是开发人员有更多partition的控制权。

1）多次读取一个消息；

2）只消费一个partition中的部分消息；

3）使用事务来保证一个消息仅被消费一次；

注意：使用此API，partition/offset/broker/leader对你不再透明，需要自己管理，需做大量的额外工作：

1）必须在应用程序中跟踪offset，从而确定下一条应该消费哪条消息；

2）应用程序需要通过程序获知每个partition的leader是谁；

3）需要处理leader的变更使用simpleConsumer API的一般流程如下：

1⃣️查好到一个活着的broker，并且找出每个partition的leader；

2⃣️找出每个partition的follower；

3⃣️定义好请求，该请求应该能描述应用程序需要那些数据；

4⃣️fetch数据；

5⃣️识别leader的变化，并对之作出必要的相应；

2:consumer group

kafka的分配单位是partition，每个consumer都属于一个group，一个partition只能被同一个group内的一个consumer所消费（保障了一个消息只能被group内的一个consumer所消费），但是多个group可以同时消费这个partition。即partition和group是一对多的关系。

kafka的设计目标之一就是同时实现离线处理和实时处理，根据这一特性，可以使用spark/Storm这些实时处理系统对消息在线处理，同时使用Hadoop批处理系统进行离线处理，还可以将数据备份到另一个数据中心，只需要保证这三者属于不同的consumer group。

3:消费方式

consumer采用pull模式从broker中读取数据。

push模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式可以根据consumer的消费能力以适当的速率消费消息。

对于kafka而言，pull模式更合适，可简化broker的设计，consumer可自主控制消费消息的速率，同时consumer可以自己控制消费方式--即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。

4:consumer delivery guarantee

实际使用中应用程序并非在consumer读取完数据就结束了，而是要进一步处理：

1）读完消息先commit再处理消息；这种模式，如果consumer在commit后还每来得及处理消息就crash了，下次重新开始工作后就无法读到刚刚已提交而未处理的消息，对应 at most once；

2）读完消息先处理再commit；这种模式，如果处理完消息之后commit之前consumer crash了，下次重新开始工作还会处理刚刚未commit的消息，实际上该消息已经处理过了，对应at least once。

3）如果一定要做到exactly once，就需要协调offset和实际操作的输出。经典做法是引入两阶段提交，如果能让offset和操作输入存在同一个地方，会更简洁和通用，这种方式可能更好，因为许多输出系统可能不支持两阶段提交。比如，consumer拿到数据后可能把数据放到HDFS，如果把最新的offset和数据本身一起写到HDFS，那就可以保证数据的输出和offset的更新要么都完成，要么都不完成，间接实现Exactly once。

（high-level API，offset存于zookeeper中，无法存于HDFS；而simpleConsumer API的offset自己维护，可存于HDFS中）

5:consumer rebalance

当有consumer加入或退出/以及partition的改变时会出发rebalance。consumer rebalance算法如下：

1）将目标topic下的所有partition排序，存于PT；

2）对某consumer group下所有consumer排序，存于CG，第i个consumer记为Ci；

3）N=size(PT)/size(CG),向上取整；

4）解除Ci对原来分配的partition的消费权（i从0开始）；

5）将第i\*N到（i+1)\*N-1个partition分配给Ci；

在0.8.\*版本后，为了保证整个consumer group的一致性，当一个consumer触发了rebalance时，该group内的其他所有consumer也触发rebalance，导致以下几个问题：

1）herd effect：任何broker活着consumer的增减都会触发所有的consumer的rebalance；

2）split brain：每个consumer分别单独通过zookeeper判断哪些broker和consumer当掉了，那么不同consumer在同一时刻从zookeeper看到的view就可能不一样，这是由zookeeper的特性决定的，这就会造成不正确的rebalance尝试；

3）调整结果不可控：所有consumer都并不直到其他consumer的rebalance是否成功，这可能会导致kafka工作在一个不正确的状态。

上述问题，将在0.9.\*版本开始使用中心coordinator来控制consumer rebalance。

[kafka]: https://km.sankuai.com/page/88333480
[kafka 1]: https://km.sankuai.com/page/96956177
[https_blog.csdn.net_laojiaqi_article_details_79034798]: https://blog.csdn.net/laojiaqi/article/details/79034798