kafka分区和副本机制_kafka分区数为1

目录

1. 分区和副本机制

1.1 生产者分区写入策略

 1.1.1 轮询策略

1.1.2 随机策略（不用）

1.1.3 按key分配策略

1.1.4 乱序问题

1.1.5 自定义分区策略

1.2 消费者组Rebalance机制

1.2.1 Rebalance再均衡

1.2.2 Rebalance的不良影响

1.3 消费者分区分配策略

1.3.1 Range范围分配策略

1.3.2 RoundRobin轮询策略

1.3.3 Stricky粘性分配策略

1.4 副本机制

1.4.1 producer的ACKs参数

1.4.2 acks配置为0

1.4.3 acks配置为1

1.4.4 acks配置为-1或者all

---

 Kafka入门及进阶汇总 

1. 分区和副本机制

1.1 生产者分区写入策略

生产者写入消息到topic，Kafka将依据不同的策略将数据分配到不同的分区中

1. 轮询分区策略
2. 随机分区策略
3. 按key分区分配策略
4. 自定义分区策略

 1.1.1 轮询策略

•  默认的策略，也是使用最多的策略，可以最大限度保证所有消息平均分配到一个分区
•  如果在生产消息时，key为null，则使用轮询算法均衡地分配分区

1.1.2 随机策略（不用）

随机策略，每次都随机地将消息分配到每个分区。在较早的版本，默认的分区策略就是随机策略，也是为了将消息均衡地写入到每个分区。但后续轮询策略表现更佳，所以基本上很少会使用随机策略。

1.1.3 按key分配策略

按key分配策略，有可能会出现「数据倾斜」，例如：某个key包含了大量的数据，因为key值一样，所有所有的数据将都分配到一个分区中，造成该分区的消息数量远大于其他的分区。

1.1.4 乱序问题

轮询策略、随机策略都会导致一个问题，生产到Kafka中的数据是乱序存储的。而按key分区可以一定程度上实现数据有序存储——也就是局部有序，但这又可能会导致数据倾斜，所以在实际生产环境中要结合实际情况来做取舍。

1.1.5 自定义分区策略

实现步骤：

1 创建自定义分区器

public class KeyWithRandomPartitioner implements Partitioner {
 
 
    private Random r;
 
 
    @Override
 
    public void configure(Map<String, ?> configs) {
 
        r = new Random();
 
    }
 
 
    @Override
 
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
 
        // cluster.partitionCountForTopic 表示获取指定topic的分区数量
 
        return r.nextInt(1000) % cluster.partitionCountForTopic(topic);
 
    }
 
 
    @Override
 
    public void close() {
 
    }
 
}

2 在Kafka生产者配置中，自定使用自定义分区器的类名

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

1.2 消费者组Rebalance机制

1.2.1 Rebalance再均衡

Kafka中的Rebalance称之为再均衡，是Kafka中确保Consumer group下所有的consumer如何达成一致，分配订阅的topic的每个分区的机制。

Rebalance触发的时机有：

1 消费者组中consumer的个数发生变化。例如：有新的consumer加入到消费者组，或者是某个consumer停止了。

2 订阅的topic个数发生变化

消费者可以订阅多个主题，假设当前的消费者组订阅了三个主题，但有一个主题突然被删除了，此时也需要发生再均衡。

3 订阅的topic分区数发生变化

1.2.2 Rebalance的不良影响

• 发生Rebalance时，consumer group下的所有consumer都会协调在一起共同参与，Kafka使用分配策略尽可能达到最公平的分配
• Rebalance过程会对consumer group产生非常严重的影响，Rebalance的过程中所有的消费者都将停止工作，直到Rebalance完成

1.3 消费者分区分配策略

1.3.1 Range范围分配策略

Range范围分配策略是Kafka默认的分配策略，它可以确保每个消费者消费的分区数量是均衡的。

注意：Rangle范围分配策略是针对每个Topic的。

配置

配置消费者的partition.assignment.strategy为org.apache.kafka.clients.consumer.RangeAssignor。

算法公式

n = 分区数量 / 消费者数量

m = 分区数量 % 消费者数量

前m个消费者消费n+1个

剩余消费者消费n个

 

1.3.2 RoundRobin轮询策略

RoundRobinAssignor轮询策略是将消费组内所有消费者以及消费者所订阅的所有topic的partition按照字典序排序（topic和分区的hashcode进行排序），然后通过轮询方式逐个将分区以此分配给每个消费者。

配置

配置消费者的partition.assignment.strategy为org.apache.kafka.clients.consumer.RoundRobinAssignor。

1.3.3 Stricky粘性分配策略

从Kafka 0.11.x开始，引入此类分配策略。主要目的：

1. 分区分配尽可能均匀
2. 在发生rebalance的时候，分区的分配尽可能与上一次分配保持相同

没有发生rebalance时，Striky粘性分配策略和RoundRobin分配策略类似。

上面如果consumer2崩溃了，此时需要进行rebalance。如果是Range分配和轮询分配都会重新进行分配，例如：

通过上图，我们发现，consumer0和consumer1原来消费的分区大多发生了改变。接下来我们再来看下粘性分配策略。

我们发现，Striky粘性分配策略，保留rebalance之前的分配结果。这样，只是将原先consumer2负责的两个分区再均匀分配给consumer0、consumer1。这样可以明显减少系统资源的浪费，例如：之前consumer0、consumer1之前正在消费某几个分区，但由于rebalance发生，导致consumer0、consumer1需要重新消费之前正在处理的分区，导致不必要的系统开销。（例如：某个事务正在进行就必须要取消了）

1.4 副本机制

副本的目的就是冗余备份，当某个Broker上的分区数据丢失时，依然可以保障数据可用。因为在其他的Broker上的副本是可用的。

1.4.1 producer的ACKs参数

对副本关系较大的就是，producer配置的acks参数了,acks参数表示当生产者生产消息的时候，写入到副本的要求严格程度。它决定了生产者如何在性能和可靠性之间做取舍。

配置：

Properties props = new Properties();
 
props.put("bootstrap.servers", "node1.itcast.cn:9092");
 
props.put("acks", "all");
 
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

1.4.2 acks配置为0

ACK为0，基准测试：

bin/kafka-producer-perf-test.sh --topic benchmark --num-records 5000000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092 acks=0

测试结果：

1.4.3 acks配置为1

当生产者的ACK配置为1时，生产者会等待leader副本确认接收后，才会发送下一条数据，性能中等。

1.4.4 acks配置为-1或者all

bin/kafka-producer-perf-test.sh --topic benchmark --num-records 5000000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092 acks=all