2018-08-26

2018-08-26 本文已影响0人 frmark

Kafka的基本概念

Broker

Kafka集群中包含多个服务器，其中每个服务器称为一个broker。有一点需要注意一下，添加一个新的broker到cluster中的时候，并不会分配任何数据partiton到新的broker，除非有新的topic被创建，为了不创建新的topic，可以考虑使用partition re-assignment tool将已有的parititon分配到新的broker中
Producer

消息生产者，向kafka broker发送消息的客户端，producer基于record的key决定将record发送到哪个partition，默认使用key的hash，如果没有key，则使用轮询的策略

利用api创建kafka生产者有三个基本属性：
1. bootstrap.servers：属性值是一个host：port的broker列表，指定了简历初始连接的broker列表，这个列表不需要包含所有的broker，因为建立的初始连接会从相应的broker获取到集群的信息。但是建议至少包含连个broker，保证高可用。
2. key.serializer：属性值是类的名称。这个属性指定了用来序列化键值（key）的类。Kafka broker只接受字节数组，但生产者的发送消息接口允许发送任何的Java对象，因此需要将这些对象序列化成字节数组。key.serializer指定的类需要实现org.apache.kafka.common.serialization.Serializer接口，Kafka客户端包中包含了几个默认实现，例如ByteArraySerializer、StringSerializer和IntegerSerializer。
3. value.serializer：属性值是类的名称。这个属性指定了用来序列化消息记录的类
4. acks： acks控制多少个副本必须写入消息后生产者才能认为写入成功，这个参数对消息丢失可能性有很大影响。这个参数有三种取值：
  - acks=0：生产者把消息发送到broker即认为成功，不等待broker的处理结果。这种方式的吞吐最高，但也是最容易丢失消息的。
  - acks=1：生产者会在该分区的群首（leader）写入消息并返回成功后，认为消息发送成功。如果群首写入消息失败，生产者会收到错误响应并进行重试。这种方式能够一定程度避免消息丢失，但如果群首宕机时该消息没有复制到其他副本，那么该消息还是会丢失。另外，如果我们使用同步方式来发送，延迟会比前一种方式大大增加（至少增加一个网络往返时间）；如果使用异步方式，应用感知不到延迟，吞吐量则会受异步正在发送中的数量限制。
  - acks=all：生产者会等待所有副本成功写入该消息，这种方式是最安全的，能够保证消息不丢失，但是延迟也是最大的。
5. 当生产者发送消息收到一个可恢复异常时，会进行重试，这个参数指定了重试的次数。在实际情况中，这个参数需要结合retry.backoff.ms（重试等待间隔）来使用，建议总的重试时间比集群重新选举群首的时间长，这样可以避免生产者过早结束重试导致失败。
```
private Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "broker1:9092,broker2:9092");
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

producer = new KafkaProducer<String, String>(kafkaProps);
```
producer创建完成后，有三种发送消息的方式：
- Fire-and-forget(即发即弃): 发送消息给服务器, 然而并不关心消息是否成功达到.大部分情况下, 它将成功达到, 因为 Kafka 是高可用的, 并且生产者会自动重试发送消息.不管怎样,使用这种方式有些消息可能会丢失.
```
ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry", "Precision Products", "France"); 
try {
    // 此方法返回 RecordMetadata, 但是这里忽略了返回值, 无法知道消息是否发送成功
    // 生产环境一般不适用此种方式
    producer.send(record); 
} catch (Exception e) {
    // SerializationException: 如果序列化失败
    // BufferExhaustedException: buffer 满了
    // TimeoutException
    // InterruptException: 发送线程被中断
    e.printStackTrace(); 
}
```
- Synchronous send(同步发送): 发送消息后, send() 方法返回一个 Future 对象, 使用 get() 方法在 future 上等待, 以此来判断 send() 是否成功.获取写入的记录的metadata，如topic、partition 和 offset
```
ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
try {
    // 发送成功, 可以获得一个 RecordMetadata 对象
    producer.send(record)
            // 等待应答
            .get(); 
} catch (Exception e) {
    // 发送失败
    e.printStackTrace(); 
}
```
  大部分情况下,我们不需要回复–Kafka 返回写入的记录的 topic、partition 和 offset,通常发送端是不需要这些的.另外,我们可能需要知道什么时候发送消息失败,所以我们可以抛出一个异常,记录错误信息或者写入错误文件用于后面的分析.不能通过重试被解决.比如, message size too large(消息大小太大),在这些情况中, KakfaProducer 将不会尝试重试, 并立即返回异常.
- Asynchronous send(异步发送): 使用一个 callback function(回调方法)调用 send() 方法, 当从 Kafka broker 接收到相应的时候会触此回调方法.
```
private class DemoProducerCallback implements Callback { 
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
     if (e != null) {//当Kafka返回异常时，异常值不为null
         e.printStackTrace(); 
        }
    }
}
ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry", "Biomedical Materials", "USA"); 
producer.send(record, new DemoProducerCallback()); 
```

Consumer（每个consumer group是一个订阅者，为每个topic partiton维护一个offset，每个consumer自己也会维护一个offset）

消息消费者，每个consumer属于一个特定的consumer group（可为每个consumer指定group name，若不指定group name则属于默认的group）。同一topic的一条消息只能被同一个consumer group内的一个consumer消费，但多个consumer group可同时消费这一消息。Consumer Group中的每个Consumer读取Topic的一个或多个Partitions，并且是唯一的Consumer；如果Consumer group中所有consumer总线程大于partitions数量，则会出现空闲情况。这样可以做到负载均衡，也可以实现顺序消费（group中只有一个consumer）。每个consumer group维护了每个topic partition的offset。

为了保证顺序消费，每个message只能发送到一个consumer中。否则效率很低，需要等到所有消费者消费完才能发送下一个message，显然是不合理的。同时对于topic中parition的消费如果是异步的就很难保证顺序性。目前许多消息系统经常使用‘独占消费’的方式消费。例如topic中的parition只能由特定一个消费者消费，官网明确kafka智能保证一个parition中的消息的有序性，不能保证topic中不同parition的有序性
如果所有的consumer都在一个consumer group中，就像传统的队列一样。如果所有的consumer都在不同的consumer group中就像发布订阅模式一样，所有的message都会广播倒所有consumers中。因此如果有很多的订阅者，kafka的性能就会降低，因为kafka需要拷贝message到所有的group中以保证顺序性
kafka consumer负载均衡：每个consumer是一个parititon的专有消费者，如果有新的consumer加入到了group中，它将获得一个共享的parititon，如果一个consumer挂了，它的partition将会被分配到其他剩余的xonsumer中
kafka灾备：consumer会将offset反馈给kafka broker当一条记录杯成功处理后。如果在发送commit offset前，consumer处理失败，其他的consumer将会继续从上次的commit offset开始处理。如果在处理完后这一条记录但还未发送commit offset时consumer发生错误，kafka记录将被重复消费。在这个情景下，kafka 实现了至少一次的消费，应该保证消息被处理时幂等的
offset 管理：kafka将offset 数据保存到一个"__consumer_offset" topic中，这个topic使用日志压缩，kafka灾备的的offset就是修改或读取此topic中的值。
consumer可以消费哪些记录？ 一条最新的记录进入之后，offset写入到log parition中，然后将该记录复制到所有的partition的followers中，最后标记"High watermark"（成功复制的最新纪录offset）。consumer 消费的是"High watermark"中的offset，未被复制的不可以被消费。
consumer和parititon的关系：对于一个group，一个consumer只能消费一个parition，如果consumer数量大于paritition的数量，有的consumer就会空闲，可以作为灾备，如果小于partiion数量，每个consumer就会消费多个parition
多线程kafka consumer ：一个consumer有多个线程，很难保证记录消费的有序性，只有在消费单条记录时间很长的时候使用，一般不建议使用。在一个进程中跑多个线程，每个线程是一个consumer，每个线程管理自己的offset。

三种消费方式

首先了解一下建立consumer的参数：

"bootstrap.servers", 指定kafka的broker

"group.id", 指定consumer group

"enable.auto.commit", 指定offset可以自动被commit 到kafka，不需要程序中显示的写

"auto.commit.interval.ms", 指定了commit offset的时间间隔

"key.serializer" and "value.serializer", are classes to be used to decode the message into bytes.

自动commit offset: parition中的一个记录被消费后自动commit offset到kafka

package com.til.kafka.consumer;  
  
import java.util.List;  
import java.util.Properties;  
  
import org.apache.kafka.clients.consumer.ConsumerRecord;  
import org.apache.kafka.clients.consumer.ConsumerRecords;  
import org.apache.kafka.clients.consumer.KafkaConsumer;  
  
import com.tb.constants.KafkaConstants;  
  
//Automatic Offset Committing  
public class AOCKafkaConsumer {  
    Properties props;  
    KafkaConsumer<String, String> consumer;  
  
    public AOCKafkaConsumer(String brokerString) {  
        props = new Properties();  
        props.put("bootstrap.servers", brokerString);  
        props.put("group.id", KafkaConstants.KAFKA_CONSUMER_GROUP);  
        props.put("enable.auto.commit", "true");  
        props.put("auto.commit.interval.ms", "1000");  
        props.put("key.deserializer", KafkaConstants.KAFKA_KEY_SERIALIZER);  
        props.put("value.deserializer", KafkaConstants.KAFKA_VALUE_SERIALIZER);  
        consumer = new KafkaConsumer<>(props);  
    }  
  
    public void subscribe(List<String> topics) {  
        consumer.subscribe(topics);  
        while (true) {  
            ConsumerRecords<String, String> records = consumer.poll(100);  
            for (ConsumerRecord<String, String> record : records)  
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());  
        }  
    }  
}  

public class KafkaConstants {  
    public static String KAFKA_BROKER_STRING =   
            "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094";  
    public static String KAFKA_KEY_SERIALIZER =   
            "org.apache.kafka.common.serialization.StringSerializer";  
    public static String KAFKA_VALUE_SERIALIZER =   
            "org.apache.kafka.common.serialization.StringSerializer";  
    public static String KAFKA_TOPIC = "TEST-1";  
    public static String KAFKA_CONSUMER_GROUP = "TEST";  
}

手动commit offset 到kafka：手动控制commit offset到kafka

import java.util.List;  
import java.util.Properties;  
  
import org.apache.kafka.clients.consumer.ConsumerRecord;  
import org.apache.kafka.clients.consumer.ConsumerRecords;  
import org.apache.kafka.clients.consumer.KafkaConsumer;  
  
import com.tb.constants.KafkaConstants;  
  
//Manual Offset Control  
public class MOCKafkaConsumer {  
    Properties props;  
    KafkaConsumer<String, String> consumer;  
  
    public MOCKafkaConsumer(String brokerString) {  
        props = new Properties();  
        props.put("bootstrap.servers", brokerString);  
        props.put("group.id", KafkaConstants.KAFKA_CONSUMER_GROUP);  
        props.put("enable.auto.commit", "false");  
        props.put("key.deserializer", KafkaConstants.KAFKA_KEY_SERIALIZER);  
        props.put("value.deserializer", KafkaConstants.KAFKA_VALUE_SERIALIZER);  
        consumer = new KafkaConsumer<>(props);  
    }  
  
    public void subscribe(List<String> topics) {  
        consumer.subscribe(topics);  
        while (true) {  
            ConsumerRecords<String, String> records = consumer.poll(100);  
            for (ConsumerRecord<String, String> record : records)  
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());  
            // This line of code manually commits offset to kafka  
            consumer.commitSync();  
        }  
    }  
}

手动分配一个consumer给一个partition：可以手工分配给特定的分区，在这种类型的使用者中，我们可以绕过使用者组的概念，并将使用者分配给特定的分区。

import java.util.Arrays;  
import java.util.List;  
import java.util.Map;  
import java.util.Map.Entry;  
import java.util.Properties;  
import java.util.Set;  
import java.util.stream.Collectors;  
  
import org.apache.kafka.clients.consumer.KafkaConsumer;  
import org.apache.kafka.common.TopicPartition;  
  
import com.tb.constants.KafkaConstants;  
  
//Manual Partition Assignment  
public class MPAKafkaConsumer {  
    private Properties props;  
    private KafkaConsumer<String, String> consumer;  
  
    public MPAKafkaConsumer(String brokerString) {  
  
        props = new Properties();  
        props.put("bootstrap.servers", brokerString);  
        // props.put("group.id", KafkaConstants.KAFKA_CONSUMER_GROUP);  
        props.put("enable.auto.commit", "false");  
        props.put("key.deserializer", KafkaConstants.KAFKA_KEY_SERIALIZER);  
        props.put("value.deserializer", KafkaConstants.KAFKA_VALUE_SERIALIZER);  
        consumer = new KafkaConsumer<>(props);  
  
    }  
  
    public void subscribe(List<TopicPartition> topicsPartions) {  
        consumer.assign(topicsPartions);  
  
    }  
  
}  

// consumer的构建和测试
import java.util.Arrays;  
  
import org.apache.kafka.common.TopicPartition;  
  
import com.tb.constants.KafkaConstants;  
import com.til.kafka.consumer.MPAKafkaConsumer;  
  
public class App {  
    public static void main(String[] args) {  
  
        // Partitions to which a consumer has to assign  
        TopicPartition partition = new   
                TopicPartition(KafkaConstants.KAFKA_TOPIC, 0);  
  
        // This will start a consumer in new thread  
        new Thread(new Runnable() {  
            @Override  
            public void run() {  
                MPAKafkaConsumer mpaKafkaConsumer =   
                        new MPAKafkaConsumer(KafkaConstants.KAFKA_BROKER_STRING);  
                  
                mpaKafkaConsumer.subscribe(Arrays.asList(partition));  
  
            }  
        }).start();  
    }  
}

Topic（复制／灾备／并行化）

可以理解为一个MQ消息队列的名字。每条发布到Kafka集群的消息都有一个类别，这个类别被称为topic。（物理上不同topic的消息分开存储，逻辑上一个topic的消息虽然保存于一个或多个broker上但用户只需指定消息的topic即可生产或消费数据而不必关心数据存于何处）。

每个topic都有一个Log（topic在硬盘上的存储），每个Log被分为多个pritions和segments。在硬盘上表现为多个文件。
Partition：

parition是物理上的概念，每个topic包含一个或多个partition，创建topic时可指定parition数量。每个partition对应于一个文件夹，该文件夹下存储该partition的数据和索引文件。为了实现扩展性，一个非常大的topic可以分布到多个 broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。也就是说，一个topic在集群中可以有多个partition，那么分区的策略是什么？(消息发送到哪个分区上，有两种基本的策略，一是采用Key Hash算法，一是采用Round Robin算法)

patition的备份数：可以配置parititon的备份数量，每个parition都有一个leader server和0到多个follower servers，其中leader server处理一个parition中的所有的读和写（和想的不太一样）。follower 复制leader，在leader挂掉后进行替换，Kafka还使用分区在组内进行并行消费者处理。Kafka在Kafka集群中的服务器上分发主题日志分区。每个服务器通过共享分区leader来处理其数据和请求的共享（不太懂）。
zookeeper

用来管理集群，协调broker／cluster的拓扑结构，管理集群中哪些broker是新增的，哪些已经挂掉了，新增了一个topic还是移除了一个topic，同时用来Broker topic partition中leader的选择。

Kafka的数据存储

主要接收topic中partition数据的存储，partition是以文件夹的形式存在具体的borker本机上（为了效率，并不依赖hdfs，自己维护多份数据）

segment文件的组成

对于一个partition（在Broker中以文件夹的形式存在），里面又有很多大小相等的segment数据文件（这个文件具体大小可以在config/server.properties中进行设置），这种特性可以方便old segment file的快速删除。
- segment file 组成：由2部分组成，分别为index file和data file，这两个文件是一一对应的，后缀”.index”和”.log”分别表示索引文件和数据文件；其中index文件结构很简单,每一行都是一个key,value对
  key 是消息的序号offset，value 是消息的物理位置偏移量.
- segment file 命名规则：partition的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset,ofsset的数值最大为64位（long类型），20位数字字符长度，没有数字用0填充。如下图所示：
查找：给定一个offset，查找message。过程如下：根据segment文件的命名，进行二分查找，找到对应的index和log文件，然后进入index 顺序查找到小于或等于offset的key（为了保证快速查找使用稀疏索引），拿到该index文件中offset对应的index，在log文件中顺序查找到需要查找的offset的message。

kafka日志清理

有两种策略：

一种是上面的cleanupLogs根据时间或大小策略（粗粒度）
还有一种是针对每个key的日志删除策略（细粒度）即LogCleaner方式，清理不包括activeSegment（即使超时），如果消息没有key，那只能采用第一种清理策略了。

日志压缩保证了：

任何消费者如果能够赶上Log的Head部分，它就会看到写入的每条消息，这些消息都是顺序递增（中间不会间断）的offset
总是维持消息的有序性，压缩并不会对消息进行重新排序，而是移除一些消息
每条消息的offset永远不会被改变，它是日志文件标识位置的永久编号
读取/消费时如果从最开始的offset=0开始，那么至少可以看到所有记录按照它们写入的顺序得到的最终状态（状态指的是value，相同key不同value，最终的状态以最新的value为准）：因为这种场景下写入顺序和读取顺序是一致的，写入时和读取时offset都是不断递增。举例写入key1的value在offset=1和offst=5的值分别是v1和v2，那么读取到offset=1时，最终的状态（value值）是v1，读取到offset=5时，最终状态是v2（不能指望说读取到offset=1时就要求状态是v2）

2018-08-26

Kafka的基本概念

三种消费方式

Kafka的数据存储

kafka日志清理

猜你喜欢

热点阅读