Kafka

系统解耦、异步通信、削峰填谷；kafka还提供了其他消息中间件难以实现的 消息顺序性保障 及回溯消费；

kafka可以把消息持久化到磁盘；kafka的消息持久化功能和多副本机制，可以使它作为长期的数据存储系统来使用；

kafka为各个流式处理框架提供了 可靠的数据来源 和 流式处理类库（窗口、连接、聚合等操作）；

生产者，负责创建消息，将消息投递到kafka；

消费者，连接到kafka上并接收消息，进而进行相应的业务逻辑处理；

服务节点（服务进程），负责接收和处理客户端发送过来的请求，以及对消息进行持久化；

发布订阅的对象，kafka中的消息以Topic为单位进行归类；Topic是一个逻辑概念；

分区，将每个主题划分成多个分区，每个分区是一组有序的消息日志（可看作可追加的日志文件）；

生产者生成的消息，只会发给主题的一个分区；（例如，双分区的主题，消息要么在分区0中，要么在分区1中）

offset

Replica

AR（Assigned Replicas）

ISR（In-Sync Replicas）

OSR（Out-of-Sync Replicas）

AR = ISR + OSR;

生产者端

Broker端

消费者端

Broker端

分类

评价指标

kafka性能结果

CPU资源充足：
Producer端完成的压缩，需要Producer运行机器上的CPU资源很充足（因为压缩需要消耗CPU）；

带宽资源有限：
消息压缩后，能够很好的节省网络传输带宽（很多生产环境都可能遭遇带宽被打满的情况）；

public interface ProducerInterceptor<K, V> extends Configurable {
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
    public void onAcknowledgement(RecordMetadata metadata, Exception exception);
    public void close();
}

public class ProducerInterceptorPrefix implements ProducerInterceptor<String,String>{
    private volatile long sendSuccess = 0;
    private volatile long sendFailure = 0;
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifiedValue = "prefix1-" + record.value();
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), modifiedValue,
            record.headers());
    }
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (e == null) {
            sendSuccess++;
        } else {
            sendFailure ++;
        }
    }
    @Override
    public void close() {
        double successRatio = (double)sendSuccess / (sendFailure + sendSuccess);
        System.out.println("[INFO] 发送成功率=" + String.format("%f", successRatio * 100) + "%");
    }
}

properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,  ProducerInterceptorPrefix.class.getName());

public interface Serializer<T> extends Closeable {
    void configure(Map<String, ?> configs, boolean isKey);
    byte[] serialize(String topic, T data);
    default byte[] serialize(String topic, Headers headers, T data) {
        return serialize(topic, data);
    }
    @Override
    void close();
}

为消息分配分区；

提供负载均衡能力，可以实现系统的高伸缩性（Scalability），可以通过添加新的节点来增加系统吞吐量；

不同的分区可以被放到不同节点的机器上，数据的读写操作的粒度也是分区，这样每个节点的机器都能
独立执行各自分区的读写请求；

public interface Partitioner extends Configurable, Closeable {
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
    void close();
}

轮询策略

随机策略

按消息键保序策略

其他分区策略

// org.apache.kafka.clients.producer.internals.DefaultPartitioner

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    int numPartitions = partitions.size();
    if (keyBytes == null) {
        int nextValue = this.nextValue(topic);
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
        if (availablePartitions.size() > 0) {
            int part = Utils.toPositive(nextValue) % availablePartitions.size();
            return ((PartitionInfo)availablePartitions.get(part)).partition();
        } else {
            return Utils.toPositive(nextValue) % numPartitions;
        }
    } else {
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

public class DemoPartitioner implements Partitioner {
    private final AtomicInteger counter = new AtomicInteger(0);
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (null == keyBytes) {
            return counter.getAndIncrement() % numPartitions;
        } else
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }
    @Override public void close() {}
    @Override public void configure(Map<String, ?> configs) {}
}

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DemoPartitioner.class.getName());

KafkaConsumer是非线程安全的（单线程架构），其中定义了一个acquire()方法，用来检测当前是否只有一个线程在操作；
如果有其他线程操作会抛出ConcurrentModifcationException 异常；

KafkaConsumer中的每个公共方法在执行所要执行的动作之前都会调用这个acquire()方法（wakeup()方法除外）；

如果所有的消费者都属于同一个消费组，那么所有的消息都会被均衡地投递给每一个消费者；

即每条消息只会被一个消费者处理；

如果所有的消费者不属于不同的消费组，那么所有的消息都会被广播给所有的消费者；

即每条消息会被所有的消费者处理；

消费组（Consumer Group）是一个逻辑上的概念，它将消费者归为一类，每一个消费者只属于一个消费组；
Consumer Group 是 Kafka 提供的可扩展且具有容错性的消费者机制；

每个消费组都会有一个固定的名称，消费者在进行消费前需要指定所属消费组的名称；参数group.id，默认为空字符串；

每个消费者都有一个对应的消费组，当消息发布到主题后，只会被投递给订阅它的每个消费组中的一个消费者；

消费组的模型可以让整体的消费能力具备横向伸缩性

分区分配的演变

所以可以通过增加（减少）消费者的个数来提高（降低）整体的消费能力；

理想情况下，一个Group中Consumer实例的数量应该等于该Group订阅主题的分区总数；

Consumer订阅了3个主题A、B、C；

分区数分别为1、2、3；

则该为这个Group设置6个Consumer实例较为合适；（可以小于，不能大于）

定义：
本质上是一种协议，规定了一个Group下的所有Consumer如何达成一致，来分配订阅Topic的每个分区；

触发条件

执行过程

分配策略

弊端

如何避免

消费组状态机
（State Machine）

消费者端Rebalance流程

Broker端（协调者端）
Rebalance流程

服务端主动将消息推送给消费者；

消费者主动向服务端发起请求来拉取消息；

poll()方法

seek()方法

以前版本

1.0.9版本以后

消息格式

创建时机

位移提交

删除过期消息

情况分类

处理方法

在某些应用场景下可能需要暂停某些分区的消费而先消费其他分区，当达到一定条件时再恢复消费这些分区；

KafkaConsumer提供了对消费速度进行控制的方法；

public void pause(Collection<TopicPartition> partitions)；（暂停分区消费）

public void resume(Collection<TopicPartition> partitions)；（恢复分区消费）

public Set<TopicPartition> paused()；（返回暂停的分区集合）

public void close()；（内部设定了最长等待时间 30s）

public void close(Duration timeout)；（设定关闭方法的最长执行时间）
因为有些关闭逻辑会耗费一定的时间，比如自动提交位移，关闭时还会执行一次位移提交操作；

public interface ConsumerInterceptor<K, V> extends Configurable {
    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records)；
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets)；
    public void close();
}

public class ConsumerInterceptorTTL implements ConsumerInterceptor<String, String> {
    private static final long EXPIRE_INTERVAL = 10000;
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        long now = System.currentTimeMillis();
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>();
        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
            List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
            for (ConsumerRecord<String, String> record : tpRecords) {
                if (now - record.timestamp() < EXPIRE_INTERVAL) {
                    newTpRecords.add(record);
                }
             }
             newRecords.put(tp, newTpRecords);
        }
        return new ConsumerRecords<>(newRecords);
    }
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        offsets.forEach((tp, offset) -> System.out.println(tp + ":" + offset.offset()));
    }
    @Override
    public void close() {}
    @Override
    public void configure(Map<String, ?> configs) {}
}

定义：
对 Producer和Consumer 要处理的消息提供什么样的承诺；

常见的承诺

kafka默认提供的交付可靠性保障是至少一次；

当Producer无法确定消息是否成功提交时，会选择重试（关闭Producer重试即可实现最多一次），
也就是再次发送相同的消息，即至少一次；（可能导致重复发送）

已提交的定义

有限度的持久化：
kafka不丢消息是有前提的，假设消息保存在N个Broker上，这个前提条件就是这N个Broker中至少有一个存活；

发后即忘（fire-and-forget）

同步（sync）

异步（async）

可重试的异常

不可重试的异常

Producer端

Broker端

Consumer端

示例

所有主题信息：
topic列表，具体的分区信息，有哪些副本，领导者副本是谁，ISR中有哪些副本等；

所有Broker信息：
当前有哪些运行的Broker，正在关闭中的Broker等；

所有涉及运维任务的分区：
当前正在进行Preferred的领导者选举以及分区重分配的分区列表；

例如使用一个kafka-topics.sh脚本创建一个分区数为3，副本数为3的主题 topic-partitions：

kafka集群中的Broker节点不可避免的会发生宕机或者崩溃，当分区的leader节点发生故障时，
ISR中的一个follower节点就会被选举成为新的leader节点；

例如将brokerId为2的节点重启，主题 topic-partitions 新的分布信息：

为了能有效地治理负载失衡的情况，kafka引入了优先副本机制（preferred replica）；

优先副本：
指 在AR集合列表中的第一个副本；比如上面分区0的AR列表为[1,2,0]，那么分区0的优先副本即为1；

理想情况下，优先副本就是该分区的leader副本，所以也可以称为 preferred leader；

优先副本选举：
指 通过一定的方式促使优先副本选举为leader副本，以此来促进集群的负载均衡（也称为“分区平衡”）；

自动

手动

在kafka中，当集群中的一个节点突然宕机时；如果节点上的分区是单副本的，则这些分区就不可用了；
如果节点上的分区是多副本的，并且这个节点还是leader节点，则这个节点的leader副本角色会转交到其他follower副本中；

不管怎样，这个节点上的分区副本都已经处于不可用的情况了，kafka并不会将这些失效的分区副本自动转移到其他节点；
这样的话，不仅会影响集群的负载均衡，还会影响整体服务的可靠性和可用性；

我们希望如果有节点突然故障，或者需要对节点进行有计划的下线操作时，能够将该节点的分区副本迁移到其他的可用节点；

当集群中新增节点时，只有新创建的主题分区才可能被分配到这个节点上，而之前的主题分区并不会自动分配到新增节点；
这样的话，新节点与原节点的负载会出现严重不均衡，所以也希望能够进行分区的重新分配；

kafka提供了 kafka-reassign-partitions.sh 脚本来执行分区重分配的工作，可以在集群扩容、节点失效的情况下对分区进行迁移；

在一个由3个节点（0、1、2）组成的集群中创建一个主题 topic-reassign，主题中包含4个分区和2个副本：

可以看到，主题 topic-reassign 在3个节点中都有相应的分区副本分布；现在需要下线 brokerId为1的节点；（需要迁移分区）

第一步：
创建一个JSON文件（reassign.json），文件内容为要进行分区重分配的主题：

第二步：
根据这个JSON文件和指定所要分配的Broker节点，来生成一份候选的重分配方案：

第三步：
执行重分配的动作；将生成的方案保存在一个JSON文件中（project.json），再执行分配：

第四步（可选）：
查看分区重分配的进度，只需要将第三步中的 execute 换成 verify 即可：

分区重分配对集群的性能影响很大，并且很占用额外的资源（网络、磁盘等）；

所以在实际操作中，需要降低重分配的粒度，分成多个小批次来执行；

如果要执行Broker下线，在执行重分配之前，最好将这个Broker 关闭或重启，这样它就不会是任何分区的leader节点，
减少broker间的流量复制；以此提升重分配的性能，减少对整个集群的影响；

生产者性能测试

消费者性能测试