Kafka

1、认识kafka

1.1、基本概念

Kafka是一个分布式消息平台

生产者：发送消息

消费者：接收消息

Broker：Kafka服务实例。一个或多个Broker组成了Kafka集群。负责将接收到的消息存储到磁盘

主题：每个消息都有一个指定主题，归类消息

分区(Partition)：主题分区，主题中包含多个分区。分区存在多副本机制，leader副本处理写请求，follower副本负责同步leader副本内容；HW：消费者可读取的分区的偏移量offset；LEO：下一条待写入消息的offset

AR：所有副本；ISR：同步滞后一定范围内的副本；OSR：同步严重滞后的副本

ZooKeeper：管理Kafka的元数据信息。

2、生产者

2.1、客户端开发

生产者生产的消息结构

public class ProducerRecord<K, V> {
    // 主题
    private final String topic;  
    // 分区
    private final Integer partition;
    // 消息头部
    private final Headers headers;
    private final K key;
    // 具体消息内容
    private final V value;
    // 消息时间戳
    private final Long timestamp;
}
复制代码

2.2、必要参数配置

初始化生产者实例 KafkaProducer 时的参数配置

bootstrap.servers Kafka集群所需的broker地址清单，无需列出所有地址，有自动发现机制

key.serializer value.serializer 消息发往broker之前key、value的值执行该序列化器，将其转换成字节数组，须写明序列化器的全限定名

client.id 设定KafkaProducer对应的客户端id(producer-1….)

KafkaProducer是线程安全的，可以在多个线程中共享单个实例

2.3、消息生产与发送

消息 ProducerRecord 的构造方法

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {
   ....
}
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
    this(topic, partition, timestamp, key, value, (Iterable)null);
}

public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {
    this(topic, partition, (Long)null, key, value, headers);
}

public ProducerRecord(String topic, Integer partition, K key, V value) {
    this(topic, partition, (Long)null, key, value, (Iterable)null);
}

public ProducerRecord(String topic, K key, V value) {
    this(topic, (Integer)null, (Long)null, key, value, (Iterable)null);
}

public ProducerRecord(String topic, V value) {
    this(topic, (Integer)null, (Long)null, (Object)null, value, (Iterable)null);
}
复制代码

其他类型的构造方法都是设置不需要参数为null，调用第一种构造

发送消息方式：

public Future<RecordMetadata> send(ProducerRecord<K, V> record) {  // Future：代表一个任务的声明周期
    return this.send(record, (Callback)null);
}

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return this.doSend(interceptedRecord, callback);
}
复制代码

发完即忘：

producer.send() 发完不关注结果

同步方式：

producer.send().get() 等待发送结果返回

异步方式：

producer.send(record, new Callback() {
    @override
    public void onCompletion(RecordMetadata metadata, Exception exception){....}
}
复制代码

2.4、序列化器

作用：序列化消息


序列化器可自定义

public class SerializerTest implements Serializer<KafkaDemo> {

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public byte[] serialize(String s, KafkaDemo kafkaDemo) {
        // 执行具体的序列化操作，返回字节数组
        return new byte[0];
    }

    @Override
    public void close() {

    }
}
复制代码

2.4、分区器

作用：为消息分配分区

逻辑：key不为null的时候对key进行Hash；key为null**轮询**发往各个可用分区

默认分区类：DefaultPartition 继承了Partition

2.5、生产者拦截器

作用：消息发送前做一些准备、拦截工作，可以统计一些信息或者过滤消息等

实现类：ProducerInterceptor 的onSend()方法

2.6、实现原理

2.6.1、生产者结构


RecordAccumulator：消息累加器，缓存消息。缓存区大小由buffer.memory参数控制默认32MB，生产者生产消息速度 > 消息发送给服务速度，导致生产者空间不足，send会被阻塞，max.block.ms控制最大阻塞时长，默认60s

分区1存储ProducerBatch的是一个双端队列Deque，后进先出，一个ProducerBatch包含n个ProducerRecord

生产者客户端中，BufferPool用来管理特定大小(batch.size:16KB)的ByteBuffer(缓存消息的内存空间)

元数据信息更新：变更触发的自动更新/定时metadata.max.age.ms(默认5min)自动更新

2.6.2、生产者参数

acks：指定分区中有多少个副本接收到消息，生产者才算发送成功。”0″ / “-1” / “1”(default值)

max.request.size能发送消息的最大值，默认值1MB

request.timeout.ms 发送消息等待请求响应最长时间，30s

retries、retry.backoff.ms 生产者重试次数；重试时间间隔。kafka保证分区消息有序

compression.type 消息压缩类型，默认为null

connections.max.idle.ms 多久之后关系闲置连接，默认9min

linger.ms 生产者发送ProducerBatch时机，Batch填满/等待时间达到linger.ms后

receive.buffer.bytes Socket接收消息缓冲区大小，32KB

send.buffer.bytes Socket发送消息缓冲区大小，128KB

3、消费者

3.1、消费者与消费组的关系

一个消费组包含一个或者多个消费者；主体中一个分区只会被消费组中一个消费者消费

消费组分区分配策略：partition.assignment.strategy

消费组名称：group.id

消息中间件支持两种消息投递模式：

1、点对点。实现方式是队列

2、发布订阅模式。生产者发送消息到主题，消费者消费主题中消息

3.2、消费者实例

3.2.1、初始化实例时参数

bootstrap.servers

group.id消费者从属的消费组id

key.deserializer、value.deserializer key和value的反序列化器

client.id 消费者对应的客户端id consumer-1…

3.2.2、订阅主题和分区

a、subscribe(Collection<String> topics, ConsumerRebalanceListener listener) ConsumerRebalanceListener listener 可选

b、subscribe(Pattern pattern, ConsumerRebalanceListener listener)ConsumerRebalanceListener listener 可选

c、assign(Collection<TopicPartition> partitions) 传入分区列表

public final class TopicPartition implements Serializable {
    private int hash = 0;
    private final int partition;
    private final String topic;
}
复制代码

使用 List<PartitionInfo> pattitionFor(String topic)方法获取主题下分区列表

a、b支持消费者自动再均衡能力，c不支持消费者自动再均衡能力

TopicPartition结构：

public class PartitionInfo {
    private final String topic;            // 主题名
    private final int partition;           // 分区号
    private final Node leader;             // leader副本
    private final Node[] replicas;         //AR集合
    private final Node[] inSyncReplicas;   // ISR集合
    private final Node[] offlineReplicas;  // OSR集合
}
复制代码

自动再均衡能力：多个消费者情况下根据分区分配策略自动分配各个消费者与分区的关系
复制代码

ByteBuffer的用法？？？

3.2.3、消息消费

消费模式：拉，poll()。消费者主动想服务器发起请求拉取消息

// 拉取消息，timeout超时时间设置
public ConsumerRecords<K, V> poll(Duration timeout) {
    return this.poll(timeout.toMillis(), true);
}

// 一次拉取操作返回的消息集
public class ConsumerRecords<K, V> implements Iterable<ConsumerRecord<K, V>> {
    public static final ConsumerRecords<Object, Object> EMPTY;
    private final Map<TopicPartition, List<ConsumerRecord<K, V>>> records;  // 分区，消息集 Map
  
    // 循环遍历消息集内部消息
    public Iterator<ConsumerRecord<K, V>> iterator() {}
    // 获取消息集包含的分区列表
    public Set<TopicPartition> partitions() {return Collections.unmodifiableSet(this.records.keySet());}
    // 获取消息集中指定分区的消息
    public List<ConsumerRecord<K, V>> records(TopicPartition partition){}
    // 获取消息集中指定主题下消息
    public Iterable<ConsumerRecord<K, V>> records(String topic){}
}

// 具体的消息结构，与ProducerRecord对应
public class ConsumerRecord<K, V> {
    public static final long NO_TIMESTAMP = -1L;
    public static final int NULL_SIZE = -1;
    public static final int NULL_CHECKSUM = -1;
    private final String topic;
    private final int partition;
    private final long offset;
    private final long timestamp;
    private final TimestampType timestampType;  // createTime：消息创建时间；LogAppendTime：消息追加到日志时间
    private final int serializedKeySize;
    private final int serializedValueSize;
    private final Headers headers;
    private final K key;
    private final V value;
    private volatile Long checksum;
}
复制代码