Kafka

kafka官网：https://kafka.apache.org/ (opens new window)

Kafka是最初由Linkedin公司开发，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。

是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统。

它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等。

一、Kafka的使用场景

• 日志收集：一个公司可以用Kafka收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。

• 消息系统：解耦、生产者和消费者、缓存消息等。

• 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

• 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

  总结：

  在大数据场景、日志收集业务时，使用kafka优势明显；而对于一般的业务场景，还是使用RocketMQ，毕竟阿里使用很多年，开发扩展了很多功能。

二、Kafka基本概念

kafka是一个分布式的，分区的消息服务(官方称之为commit log)。它提供一个消息系统应该具备的功能，但是却有着独特的设计。可以这样来说，Kafka借鉴了JMS规范的思想，但是却并没有完全遵循JMS规范。

首先，让我们来看一下基础的消息(Message)相关术语：

名称	解释
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内部消息是有序的

因此，从一个较高的层面上来看，producer通过网络发送消息到Kafka集群，然后consumer来进行消费，如下图：

主题Topic

可以理解Topic是一个类别的名称，同类消息发送到同一个Topic下面。

Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic。

生产者Producer

写入方式：

producer 采用 push 模式将消息发布到 broker，每条消息都被 append 追加到 patition 中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。

producer发布消息选择路由：

producer 发送消息到 broker 的topic中时，会根据分区算法选择将其存储到哪一个 partition。

其路由机制为：

1. 指定了 patition，则直接使用；
2. 未指定 patition 但指定 key，通过对 key 的值进行hash， 选出一个 patition；
3. patition 和 key 都未指定，使用轮询选出一个 patition。

当producer发布一个消息到某个指定的Topic，这个Topic如果不存在，就自动创建。

消费者Consumer

传统的消息传递模式有2种：队列( queue) 和发布订阅（publish-subscribe）

1. 单播消费——队列( queue)

一条消息只能被某一个消费者消费的模式。

针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性，一个partition同一个时刻在一个consumer group中，只能有一个consumer instance在消费，从而保证消费顺序。

2. 多播消费——发布订阅（publish-subscribe）

消息会被广播给所有的consumer。

所有的consumer都有着自己唯一的consumer group。

消费顺序

Kafka只在partition的范围内保证消息消费的局部顺序性。而不能在同一个topic中的多个partition中保证总的消费顺序性。

如果有在总体上保证消费顺序的需求，那么我们可以通过将topic的partition数量设置为1，将consumer group中的consumer instance数量也设置为1，但是这样会影响性能，所以kafka的顺序消费很少用。

消费者组

每个Consumer属于一个特定的Consumer Group。

一条消息可以被多个不同的Consumer Group消费；但是一个Consumer Group中只能有一个Consumer能够消费该消息。

分区Partition

Partition是一个有序的message序列。对于每一个Topic，可以有多个分区。

Q：为什么要对Topic下数据进行分区存储？

1. commit log文件会受到所在机器的文件系统大小的限制，分区之后可以将不同的分区放在不同的机器上，相当于对数据做了分布式存储，理论上一个topic可以处理任意数量的数据。

2. 为了提高并行度。

commit log 日志文件

每个partition，都对应一个commit log文件。对于每一个Topic，下面可以有多个分区日志文件（commit log）:

这些message按顺序添加到一个叫做commit log的文件中——该文件会存储在一个文件夹下，以topic名称+分区号命名。

kafka的消息message被消费后，不会立即进行删除操作。message存储在commit log文件中，默认只保留7天，7天之后，不管有无消费，都会进行删除。

log.retention.hours=168（默认7天），表示每个日志文件删除之前保存的时间。数据保存时间对所有topic都一样。

偏移量offset

每个partition中的消息都有一个唯一的编号，称之为offset，用来唯一标示某个分区中的message。

每个consumer是基于自己在commit log中的消费进度(offset)来进行工作的。

在kafka中，消费offset由consumer自己来维护。

一般情况下我们按照顺序逐条消费commit log中的消息，当然我可以通过指定offset来重复消费某些消息，或者跳过某些消息。

这意味kafka中的consumer对集群的影响是非常小的，添加一个或者减少一个consumer，对于集群或者其他consumer来说，都是没有影响的，因为每个consumer维护各自的消费offset。

查看消费组的消费偏移量

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.65.60:9092 --describe --group testGroup

• **current-offset：**当前消费组的已消费偏移量
• **log-end-offset：**topic对应分区消息的结束偏移量(HW)
• **lag：**当前消费组未消费的消息数

可以这么来理解Topic，Partition和Broker

一个topic，代表逻辑上的一个业务数据集，比如按数据库里不同表的数据操作消息区分放入不同topic，订单相关操作消息放入订单topic，用户相关操作消息放入用户topic。

对于大型网站来说，后端数据都是海量的，订单消息很可能是非常巨量的，比如有几百个G甚至达到TB级别，如果把这么多数据都放在一台机器上可定会有容量限制问题，那么就可以在topic内部划分多个partition来分片存储数据。

不同的partition可以位于不同的机器上，每台机器上都运行一个Kafka的进程Broker。

kafka集群

对于kafka来说，一个单独的broker意味着kafka集群中只有一个节点。

要想增加kafka集群中的节点数量，只需要多启动几个broker实例即可。

kafka将很多集群关键信息记录在zookeeper里，保证自己的无状态，从而在水平扩容时非常方便。

kafka leader节点

与其他中间件不同，kafka的leader是基于分区Partition而区分的。

针对每个partition，都有一个broker起到“leader”的作用，0个或多个其他的broker作为“follwers”的作用。

同一个主题的不同分区leader副本一般不一样(为了容灾)。

leader作用：leader处理所有针对这个partition的读写请求；如果这个leader失效了，其中的一个follower将会自动选举变成新的leader。

follwers作用：副本用于被动复制（replicas）备份数据；提高系统容错性。不提供读写(主要是为了保证多副本数据与消费的一致性)。

三、Kafka安装

安装前置准备

安装JDK

由于Kafka是用Scala语言开发的，运行在JVM上，因此在安装Kafka之前需要先安装JDK。

yum install java-1.8.0-openjdk* -y

安装zookeeper

kafka依赖zookeeper，所以需要先安装zookeeper

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz
tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
cd  apache-zookeeper-3.5.8-bin
cp conf/zoo_sample.cfg conf/zoo.cfg

# 启动zookeeper
bin/zkServer.sh start
bin/zkCli.sh
#查看zk的根目录相关节点
ls /

安装kafka

linux安装

docker安装

zookeeper节点数据

四、JAVA使用

生产者

引入kafka依赖

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.4.1</version>
</dependency>

生产者配置

@Bean(name="kafkaProducer")
public KafkaProducer<String,String> producer(){
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
     /*
     发送消息持久化机制参数 Default:  all
     */
    props.put(ProducerConfig.ACKS_CONFIG, "1");
     /*
    失败重试次数，发送失败会重试，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在接收者那边做好消息接收的幂等性处理
    */
    props.put(ProducerConfig.RETRIES_CONFIG, 3);
    //重试间隔设置，默认重试间隔100ms，
    props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
    //设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送性能，默认值是33554432，即32MB
    props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    /*
    kafka本地线程会从缓冲区取数据，批量发送到broker，
    设置批量发送消息的大小，默认值是16384，即16kb，就是说一个batch满了16kb就发送出去
    */
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    /*
    默认值是0，意思就是消息必须立即被发送，但这样会影响性能
    一般设置10毫秒左右，就是说这个消息发送完后会进入本地的一个batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去
    如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长
    */
    props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
    //把发送的key从字符串序列化为字节数组
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    //把发送消息value从字符串序列化为字节数组
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
}

生产者发送消息

ProducerRecord 同步阻塞方式：

@Resource(name = "kafkaProducer")
private KafkaProducer<String, String> kafkaProducer;

// jsonObj为消息的json对象串形式
// kafkaTopicNm为topic名称
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(kafkaTopicNm, jsonObj.toJSONString());
kafkaProducer.send(producerRecord);

说明：

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

1. 指定了 patition，则直接使用；
2. 未指定 patition 但指定 key，通过对 key 的 value 进行hash， 选出一个 patition；
3. patition 和 key 都未指定，使用轮询选出一个 patition。

当producer发布一个消息到某个指定的Topic，这个Topic如果不存在，就自动创建。

ProducerRecord 异步回调方式：

kafkaProducer.send(producerRecord, new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            System.err.println("发送消息失败：" + exception.getStackTrace());

        }
        if (metadata != null) {
            System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
        }
    }
});
kafkaProducer.close();

消费者

消费者配置

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
// 消费分组名
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
// 是否自动提交offset，默认就是true
/*props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 自动提交offset的间隔时间。Default: 5000 (5 seconds)
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");*/
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
/*
当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费
  latest(默认) ：只消费自己启动之后发送到主题的消息
  earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
*/
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
/*
consumer给broker发送心跳的间隔时间，broker接收到心跳如果此时有rebalance发生会通过心跳响应将
rebalance方案下发给consumer，这个时间可以稍微短一点。Default: 3000 (3 seconds)
*/
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
/*
服务端broker多久感知不到一个consumer心跳就认为他故障了，会将其踢出消费组，
对应的Partition也会被重新分配给其他consumer，默认默认Default:  45000 (45 seconds)
*/
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

//一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点。Default: 500
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
/*
如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，
会将其踢出消费组，将分区分配给别的consumer消费。Default: 300000 (5 minutes)
*/
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

消费者使用

KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
// 订阅主题
consumer.subscribe(Arrays.asList(TOPIC_NAME));
// 消费指定分区
//consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

//消息回溯消费
/*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/

//指定offset消费
/*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/

//从指定时间点开始消费

/*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
//从1小时前开始消费
long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
Map<TopicPartition, Long> map = new HashMap<>();
for (PartitionInfo par : topicPartitions) {
    map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
}
Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
    TopicPartition key = entry.getKey();
    OffsetAndTimestamp value = entry.getValue();
    if (key == null || value == null) continue;
    Long offset = value.offset();
    System.out.println("partition-" + key.partition() + "|offset-" + offset);
    System.out.println();
    //根据消费里的timestamp确定offset
    if (value != null) {
        consumer.assign(Arrays.asList(key));
        consumer.seek(key, offset);
    }
}*/

while (true) {
    /*
     * poll() API 是拉取消息的长轮询
     */
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                record.offset(), record.key(), record.value());
    }

    if (records.count() > 0) {
        // 手动同步提交offset，当前线程会阻塞直到offset提交成功
        // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
        //consumer.commitSync();

        // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
        /*consumer.commitAsync(new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if (exception != null) {
                    System.err.println("Commit failed for " + offsets);
                    System.err.println("Commit failed exception: " + exception.getStackTrace());
                }
            }
        });*/

    }
        }

解读上述代码，如下：

订阅主题

consumer.subscribe(Arrays.asList(TOPIC_NAME));

指定分区消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

消息回溯消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

指定offset消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);

指定时间消费

List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
//从1小时前开始消费
long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
Map<TopicPartition, Long> map = new HashMap<>();
for (PartitionInfo par : topicPartitions) {
    map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
}
Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
    TopicPartition key = entry.getKey();
    OffsetAndTimestamp value = entry.getValue();
    if (key == null || value == null) continue;
    Long offset = value.offset();
    System.out.println("partition-" + key.partition() + "|offset-" + offset);
    System.out.println();
    //根据消费里的timestamp确定offset
    if (value != null) {
        consumer.assign(Arrays.asList(key));
        consumer.seek(key, offset);
    }
}

指定时间进行消费，其实是先根据指定时间，找到对应timestamp对应的offset，再根据offset进行消费。

其底层消费逻辑是：从根据获得的时间timestamp，到时间索引文件（.timeindex）二分查找法找到对应的offset；再通过索引文件（.index）进行二分查找进行定位；然后去日志文件（.log）中查找具体的消息message。

五、SpringBoot整合Kafka

引入spring boot kafka依赖

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

application.yml配置

spring:
  kafka:
    bootstrap-servers: 192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094
    producer: # 生产者
      retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default-group
      enable-auto-commit: false
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
      ack-mode: manual_immediate

ack-mode:

• RECORD：当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
• BATCH：当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
• TIME：当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交。
• COUNT：当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交。
• COUNT_TIME：TIME |　COUNT　有一个条件满足时提交。
• MANUAL：当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交。（手动批量提交）
• MANUAL_IMMEDIATE：手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种。（手动单条提交）

Producer发送者

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

kafkaTemplate.send(TOPIC_NAME, 0, "key", "this is a msg");

Consumer消费者

消费者消费消息有2种方式：

单例消费：

    @KafkaListener(topics = "my-replicated-topic",groupId = "zhugeGroup")
    public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        // 手动提交offset
        ack.acknowledge();
    }

消费多个主题、指定分区：

    // concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
	@KafkaListener(groupId = "testGroup", topicPartitions = {
                @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
                @TopicPartition(topic = "topic2", partitions = "0",
                        partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        },concurrency = "6")

批量消费：

六、Kafka设计原理

Producer核心参数

发送消息持久化机制参数acks

// public static final String ACKS_CONFIG = "acks";
props.put(ProducerConfig.ACKS_CONFIG, "1");

• acks=0：表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。

• acks=1：默认值。至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。

• acks=all或acks=-1：需要等待min.insync.replicas(默认为1，推荐配置大于等于2)这个参数配置的副本个数都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。

  一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。

失败重试次数retries

// public static final String RETRIES_CONFIG = "retries";
props.put(ProducerConfig.RETRIES_CONFIG, 3)
// public static final String RETRY_BACKOFF_MS_CONFIG = "retry.backoff.ms";
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);

retries：失败重试次数，发送失败会重试，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动。所以需要在接收者那边做好消息接收的幂等性处理。

retry.backoff.ms：重试间隔设置，默认重试间隔100ms。

批量消费

// public static final String BUFFER_MEMORY_CONFIG = "buffer.memory";
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
// public static final String BATCH_SIZE_CONFIG = "batch.size";
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// public static final String LINGER_MS_CONFIG = "linger.ms";
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);

• buffer.memory：设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送性能，默认值是33554432，即32MB。

• batch.size：kafka本地线程会从缓冲区取数据，批量发送到broker，设置批量发送消息的大小，默认值是16384，即16kb，就是说一个batch满了16kb就发送出去。

• linger.ms：默认值是0，意思就是消息必须立即被发送，但这样会影响性能。一般设置10毫秒左右。

  就是说这个消息发送完后会进入本地的一个batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去；

  如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长。

Consumer核心参数

消费组

// public static final String GROUP_ID_CONFIG = "group.id";
// 消费分组名
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);

每个消费者必须指定一个消费者组；若你没有指定，kafka会默认给你生成一个组。

自动提交offset

// "enable.auto.commit"
// 是否自动提交offset，默认就是true 
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// "auto.commit.interval.ms"
// 自动提交offset的间隔时间 
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

当配置enable.auto.commit为true时，kafka消费者会默认在消费消息后，将当前消费的消息的offset提交至kafka内部topic：**__consumer_offsets**进行存储，以便下次消费时使用。

配置auto.commit.interval.ms=1000，表示1秒钟后才进行自动提交offset。此时可能会有2种极端情况：

1. 在自动提交间隔时间到了之后，消息还没有被消费者处理完，而此时由于配置了该间隔时间，offset已经自动提交了，这时若程序宕机，就会造成消息丢失问题。
2. 在自动提交间隔时间到之前，消息经过0.1秒就已经处理消费完毕，而在offset提交之前的程序运行时发生宕机，那么就会造成消息重复消费问题。

一般情况下，程序会不会设置为自动提交，而是等消息成功消费后，进行手动提交。

• 手动同步提交：consumer.commitSync();
• 手动异步提交：consumer.commitAsync(new OffsetCommitCallback() {});

批量消费条数

// "max.poll.records"
// 一次poll最大拉取消息的条数。
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);

如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点.

两次消费最大间隔时间

// "max.poll.interval.ms"
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，会将其踢出消费组，将分区分配给别的consumer消费。

心跳时间间隔

// "heartbeat.interval.ms"
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);

consumer给broker发送心跳的间隔时间。

broker接收到心跳，如果此时有Rebalance发生，会通过心跳响应将Rebalance方案下发给consumer，这个时间可以稍微短一点。

// "session.timeout.ms"
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

服务端broke感知consumer的心跳间隔时间，默认是10秒。

如果10秒后还感知不到一个consumer心跳，就认为他故障了，会将其踢出消费组，对应的Partition也会被重新分配给其他consumer。

首次消费offset设置

// "auto.offset.reset"
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费

• latest(默认) ：只消费自己启动之后发送到主题的消息

• earliest：第一次从头开始消费，以后按照消费offset记录继续消费。

  这个需要区别于consumer.seekToBeginning(每次都从头开始消费)。

kafka总控制器Controller

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为总控制器（Kafka Controller）。

它负责管理整个集群中所有分区和副本的状态：

• 当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。
• 当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
• 当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

Controller选举机制

在kafka集群启动的时候，会自动选举一台broker作为controller来管理整个集群，选举的过程是集群中每个broker都会尝试在zookeeper上创建一个 /controller 临时节点（create命令），zookeeper会保证有且仅有一个broker能创建成功，这个broker就会成为集群的总控器controller。

当这个controller角色的broker宕机了，此时zookeeper临时节点会消失，集群里其他broker会一直监听这个临时节点，发现临时节点消失了，就竞争再次创建临时节点，就是我们上面说的选举机制，zookeeper又会保证有且仅有一个broker成为新的controller。

Controller作用

具备控制器身份的broker需要比其他普通的broker多一份职责，具体细节如下：

1. 监听broker相关的变化。为Zookeeper中的/brokers/ids/节点添加BrokerChangeListener，用来处理broker增减的变化。
2. 监听topic相关的变化。为Zookeeper中的/brokers/topics节点添加TopicChangeListener，用来处理topic增减的变化；为Zookeeper中的/admin/delete_topics节点添加TopicDeletionListener，用来处理删除topic的动作。
3. 从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。对于所有topic所对应的Zookeeper中的/brokers/topics/[topic]节点添加PartitionModificationsListener，用来监听topic中的分区分配变化。
4. 更新集群的元数据信息，同步到其他普通的broker节点中。

分区Leader选举机制

副本进入ISR列表有两个条件：

1. 副本节点不能产生分区，必须能与zookeeper保持会话以及跟leader副本网络连通
2. 副本能复制leader上的所有写操作，并且不能落后太多。

与leader副本同步滞后的副本，是由 replica.lag.time.max.ms 配置决定的，超过这个时间都没有跟leader同步过的一次的副本会被移出ISR列表。

选举机制：

Controller感知到分区leader所在的broker挂了（Controller监听了很多zk节点可以感知到broker存活），

1. 参数unclean.leader.election.enable=false的前提下（默认）：

Controller会从ISR列表里挑第一个broker作为leader，第一个broker最先放进ISR列表，可能是同步数据最多的副本。

2. 如果参数unclean.leader.election.enable为true：

代表在ISR列表里所有副本都挂了的时候，可以在ISR列表以外的副本中选leader，这种设置，可以提高可用性，但是选出的新leader有可能丢失数据。

生产者send流程

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。

HW高水位

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO(log-end-offset)作为HW，consumer最多只能消费到HW所在的位置。

另外每个replica都有HW，leader和follower各自负责更新自己的HW的状态。

对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。

对于来自内部broker的读取请求，没有HW的限制。

Kafka集群副本复制原理

Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。

事实上，同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种复制方式极大的影响了吞吐率。

而异步复制方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据。

而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。

消费者消费消息的offset记录机制

每个consumer会定期将自己消费分区的offset提交给kafka内部topic：__consumer_offsets。

提交的key是consumerGroupId+topic+分区号，value就是当前offset的值。

kafka会定期清理topic里的消息，最后就保留最新的那条数据。

因为__consumer_offsets可能会接收高并发的请求，kafka默认给其分配50个分区，这样可以通过加机器的方式抗大并发。

可以通过offsets.topic.num.partitions设置该数值。

通过如下公式可以计算出consumer消费的offset要提交到__consumer_offsets的哪个分区

公式：hash(consumerGroupId) % __consumer_offsets主题的分区数

消费者Rebalance机制

Rebalance机制就是说如果消费组里的消费者数量有变化或消费的分区数有变化，kafka会重新分配消费者消费分区的关系。

比如consumer group中某个消费者挂了，此时会自动把分配给它的分区交给其他的消费者。

如果它又重启了，那么又会把一些分区重新交还给它。

注意：

rebalance只针对subscribe这种不指定分区消费的情况。如果通过assign这种消费方式指定了分区，kafka不会进行Rebanlance。

如下情况可能会触发消费者Rebalance：

1. 消费组里的Consumer增加或减少了
2. 动态给topic增加了分区
3. 消费组订阅了更多的topic

Rebalance过程中，消费者无法从kafka消费消息，这对kafka的TPS（系统的吞吐量）会有影响。

如果kafka集群内节点较多，比如数百个，那重平衡可能会耗时极多，所以应尽量避免在系统高峰期的重平衡发生。

Rebalance分区分配策略：

主要有三种rebalance的策略：range、round-robin、sticky。

Kafka 提供了消费者客户端参数partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略。默认情况为range分配策略。

假设一个主题有10个分区(0-9)，现在有三个consumer消费：

range策略就是范围分配，按照分区序号排序，假设 n＝分区数／消费者数量 = 3， m＝分区数%消费者数量 = 1，那么前 m 个消费者每个分配 n+1 个分区，后面的（消费者数量－m ）个消费者每个分配 n 个分区。

比如分区0~3给一个consumer，分区4~6给一个consumer，分区7~9给一个consumer。

round-robin策略就是轮询分配，比如分区0、3、6、9给一个consumer，分区1、4、7给一个consumer，分区2、5、8给一个consumer。

sticky策略初始时分配策略与round-robin类似，但是在Rebalance的时候，需要保证如下两个原则：

1）分区的分配要尽可能均匀 。

2）分区的分配尽可能与上次分配的保持相同。

当两者发生冲突时，第一个目标优先于第二个目标 。这样可以最大程度维持原来的分区分配的策略。

比如对于第一种range情况的分配，如果第三个consumer挂了，那么重新用sticky策略分配的结果如下：

• consumer1除了原有的0~3，会再分配一个7

• consumer2除了原有的4~6，会再分配8和9

Rebalance过程

当有消费者加入消费组时，消费者、消费组及组协调器之间会经历以下几个阶段。

第一阶段：选择组协调器

组协调器GroupCoordinator：每个consumer group都会选择一个broker作为自己的组协调器coordinator，负责监控这个消费组里的所有消费者的心跳，以及判断是否宕机，然后开启消费者rebalance。

consumer group中的每个consumer启动时会向kafka集群中的某个节点发送 FindCoordinatorRequest 请求来查找对应的组协调器GroupCoordinator，并跟其建立网络连接。

组协调器选择方式：

consumer消费的offset要提交到__consumer_offsets的哪个分区，这个分区leader对应的broker就是这个consumer group的coordinator

第二阶段：加入消费组 JOIN GROUP

在成功找到消费组所对应的 GroupCoordinator 之后就进入加入消费组的阶段，在此阶段的消费者会向 GroupCoordinator 发送 JoinGroupRequest 请求，并处理响应。然后GroupCoordinator 从一个consumer group中选择第一个加入group的consumer作为leader(消费组协调器)，把consumer group情况发送给这个leader，接着这个leader会负责制定分区方案。

第三阶段（ SYNC GROUP)

consumer leader通过给GroupCoordinator发送SyncGroupRequest，接着GroupCoordinator就把分区方案下发给各个consumer，他们会根据指定分区的leader broker进行网络连接以及消息消费。

日志分段存储及索引原理

日志分段存储

消息在分区内是分段(segment)存储，每个段的消息都存储在不一样的log文件里，这种特性方便old segment file快速被删除，kafka规定了一个段位的 log 文件最大为 1G，做这个限制目的是为了方便把 log 文件加载到内存去操作。

Kafka Broker 有一个参数log.segment.bytes，限定了每个日志段文件的大小，最大就是 1GB。

一个日志段文件满了，就自动开一个新的日志段文件来写入，避免单个文件过大，影响文件的读写性能，这个过程叫做 log rolling。

新的日志文件名称命名为最新一条数据的offset的值。

例：00000000000005367851.log表示当前的日志段文件中的第一个offset值为5367851。

日志文件

每次分段存储消息，都会存储至三个文件：

• 000000xxxxx.index
• 000000xxxxx.log
• 000000xxxxx.timeindex

消息存储文件（.log）：

消息存储文件，主要存offset和消息体，是真正存储消息的地方。

消息索引文件（.index）：

该索引文件为稀疏索引。存储指定间隔的offset索引。

kafka每次往分区发4K(可配置)，消息就会记录一条当前消息的offset到index文件。

索引文件的作用：如果要定位消息，会先根据offset在这个文件里快速定位，再去log文件里找具体消息。

消息时间索引文件（.timeindex）:

消息的发送时间索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对应的offset，存储到timeindex文件。

如果需要按照时间来定位消息的offset，会先在这个文件里根据时间查找对应的offset，再去log文件查找具体消息。

七、Kafka可视化管理工具

kafka-manager: 为了简化开发者和服务工程师维护Kafka集群的工作，雅虎yahoo构建了一个叫做Kafka管理器的基于Web工具，叫做 Kafka Manager。这个管理工具可以很容易地发现分布在集群中的哪些topic分布不均匀，或者是分区在整个集群分布不均匀的的情况。

它支持管理多个集群、选择副本、副本重新分配以及创建Topic。同时，这个管理工具也是一个非常好的可以快速浏览这个集群的工具。

安装及基本使用可参考：https://www.cnblogs.com/dadonggg/p/8205302.html (opens new window)

kafka-manager 项目地址：https://github.com/yahoo/kafka-manager (opens new window)

默认的端口是9000

功能：

1. 管理多个kafka集群
2. 便捷的检查kafka集群状态(topics,brokers,备份分布情况,分区分布情况)
3. 选择你要运行的副本
4. 基于当前分区状况进行
5. 可以选择topic配置并创建topic(0.8.1.1和0.8.2的配置不同)
6. 删除topic(只支持0.8.2以上的版本并且要在broker配置中设置delete.topic.enable=true)
7. Topic list会指明哪些topic被删除（在0.8.2以上版本适用）
8. 为已存在的topic增加分区
9. 为已存在的topic更新配置
10. 在多个topic上批量重分区
11. 在多个topic上批量重分区(可选partition broker位置)

八、kafka高性能优化

高性能集群规划

JVM参数设置

kafka是scala语言开发，运行在JVM上，需要对JVM参数合理设置。

修改bin/kafka-start-server.sh中的jvm设置，假设机器是32G内存，可以如下设置：

export KAFKA_HEAP_OPTS="-Xmx16G -Xms16G -Xmn10G -XX:MetaspaceSize=256M -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:G1HeapRegionSize=16M"

这种大内存的情况一般都要用G1垃圾收集器，因为年轻代内存比较大，用G1可以设置GC最大停顿时间，不至于一次minor gc就花费太长时间。

当然，因为像kafka，rocketmq，es这些中间件，写数据到磁盘会用到操作系统的page cache，所以JVM内存不宜分配过大，需要给操作系统的缓存留出几个G。

消息丢失问题

生产者Producer

（1）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。

性能最高，但是最容易丢消息。大数据统计报表场景，对性能要求很高，对数据丢失不敏感的情况可以用这种。

（2）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。

这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。

（3）acks=-1或all： 这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。

当然如果min.insync.replicas配置的是1，则也可能丢消息，跟acks=1情况类似。

解决方案：

消费者Consumer

如果消费这边配置的是自动提交，万一消费到数据还没处理完，就自动提交offset了。但是此时你consumer直接宕机了，未处理完的数据丢失了，下次也消费不到了。

解决方案：

消息重复消费

生产者Producer

发送消息如果配置了重试机制，比如网络抖动时间过长导致发送端发送超时，实际broker可能已经接收到消息，但生产者会重新发送消息。

解决方案：

消费者Consumer

如果消费这边配置的是自动提交，刚拉取了一批数据处理了一部分，但还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理。

解决方案：

一般消费端都是要做消费幂等处理的。

消息乱序

如果发送端配置了重试机制，kafka不会等之前那条消息完全发送成功才去发送下一条消息，这样可能会出现，发送了1，2，3条消息，第一条超时了，后面两条发送成功，再重试发送第1条消息，这时消息在broker端的顺序就是2，3，1。

所以，是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息，这样也能保证消息发送的有序。当然acks不能设置为0，防止消息丢失。

kafka保证全链路消息顺序消费，需要从发送端开始，将所有有序消息发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低。

可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。

消息积压

消息积压场景：

1. 线上有时因为发送方发送消息速度过快，或者消费方处理消息过慢，可能会导致broker积压大量未消费消息。

解决：

此种情况如果积压了上百万未消费消息需要紧急处理，可以修改消费端程序，让其将收到的消息快速转发到其他topic(可以设置很多分区)，然后再启动多个消费者同时消费新主题的不同分区。

2. 由于消息数据格式变动或消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未消费消息。

   解决：

   此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)，后面再慢慢分析死信队列里的消息处理问题。

消息回溯

如果某段时间对已消费消息计算的结果觉得有问题，可能是由于程序bug导致的计算错误，当程序bug修复后，这时可能需要对之前已消费的消息重新消费，可以指定从多久之前的消息回溯消费，这种可以用consumer的offsetsForTimes、seek等方法指定从某个offset偏移的消息开始消费。

延时队列

延时队列存储的对象是延时消息。

所谓的“延时消息”是指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费。

延时队列的使用场景有很多， 比如 ：

1. 在订单系统中， 一个用户下单之后通常有 30 分钟的时间进行支付，如果 30 分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延时队列来处理这些订单了。
2. 订单完成1小时后通知用户进行评价。

实现思路：

发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中（topic_1s，topic_5s，topic_10s，...topic_2h，这个一般不能支持任意时间段的延时）；

然后通过定时器进行轮训消费这些topic，查看消息是否到期；

如果到期就把这个消息发送到具体业务处理的topic中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了。

消息传递保障

• at most once(消费者最多收到一次消息，0--1次)：acks = 0 可以实现。
• at least once(消费者至少收到一次消息，1--多次)：ack = all 可以实现。
• exactly once(消费者刚好收到一次消息)：
  • at least once 加上消费者幂等性可以实现；
  • 还可以用kafka生产者的幂等性来实现。

kafka生产者的幂等性：

因为发送端重试导致的消息重复发送问题，kafka的幂等性可以保证重复发送的消息只接收一次，只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可，默认是false不开启。

具体实现原理：

kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker。broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查PID和Sequence Number，如果相同不会再接收。

PID：每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个PID 对用户完全是透明的。生产者如果重启则会生成新的PID。

Sequence Number：对于每个 PID，该 Producer 发送到每个 Partition 的数据都有对应的序列号，这些序列号是从0开始单调递增的。

压测分区数量

可以用kafka压测工具自己测试分区数不同，各种情况下的吞吐量。

案例：往test主题里发送一百万消息，每条设置1KB。

bin/kafka-producer-perf-test.sh --topic test --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=192.168.65.60:9092 acks=1

• throughput 用来进行限流控制，当设定的值小于 0 时不限流；当设定的值大于 0 时，发送的吞吐量大于该值时就会被阻塞一段时间。

网络上很多资料都说分区数越多吞吐量越高 ， 但从压测结果来看，分区数到达某个值吞吐量反而开始下降，实际上很多事情都会有一个临界值，当超过这个临界值之后，很多原本符合既定逻辑的走向又会变得不同。

一般情况分区数跟集群机器数量相当就差不多了。

当然吞吐量的数值和走势还会和磁盘、文件系统、 I/O调度策略等因素相关。

注意：如果分区数设置过大，比如设置10000，可能会设置不成功，后台会报错"java.io.IOException : Too many open files"。

异常中最关键的信息是“ Too many open flies”，这是一种常见的 Linux 系统错误，通常意味着文件描述符不足，它一般发生在创建线程、创建 Socket、打开文件这些场景下 。 在 Linux系统的默认设置下，这个文件描述符的个数不是很多 ，通过 ulimit -n 命令可以查看：一般默认是1024，可以将该值增大，比如：ulimit -n 65535。

kafka的事务

Kafka的事务不同于Rocketmq，Rocketmq是保障本地事务(比如数据库)与mq消息发送的事务一致性。而Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)。

一般在kafka的流式计算场景用得多一点。比如，kafka需要对一个topic里的消息做不同的流式计算处理，处理完分别发到不同的topic里，这些topic分别被不同的下游系统消费(比如hbase，redis，es等)，这种我们肯定希望系统发送到多个topic的数据保持事务一致性。

Kafka要实现类似Rocketmq的分布式事务需要额外开发功能。

kafka的事务处理可以参考官方文档 (opens new window)。

 Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("transactional.id", "my-transactional-id");
 Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
 //初始化事务
 producer.initTransactions();

 try {
     //开启事务
     producer.beginTransaction();
     for (int i = 0; i < 100; i++){
         //发到不同的主题的不同分区
         producer.send(new ProducerRecord<>("hdfs-topic", Integer.toString(i), Integer.toString(i)));
         producer.send(new ProducerRecord<>("es-topic", Integer.toString(i), Integer.toString(i)));
         producer.send(new ProducerRecord<>("redis-topic", Integer.toString(i), Integer.toString(i)));
     }
     //提交事务
     producer.commitTransaction();
 } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
     // We can't recover from these exceptions, so our only option is to close the producer and exit.
     producer.close();
 } catch (KafkaException e) {
     // For all other exceptions, just abort the transaction and try again.
     //回滚事务
     producer.abortTransaction();
 }
 producer.close();

kafka高性能的原因

• 磁盘顺序读写
• 数据传输的零拷贝
• 读写数据的batch批量处理以及压缩传输

kafka消息不能修改以及不会从文件中删除，保证了磁盘顺序读；

kafka的消息写入文件都是追加在文件末尾，不会写入文件中的某个位置(随机写)，保证了磁盘顺序写。

数据传输零拷贝原理：

九、Kafka相关资料

• 官方：

kafka官网 (opens new window)

官方—生产者配置项说明 (opens new window)

官方—消费者配置项说明 (opens new window)

• 笔记：

kafka有道云笔记01—图灵诸葛老师 (opens new window)

kafka有道云笔记02—图灵诸葛老师 (opens new window)

kafka有道云笔记03—图灵诸葛老师 (opens new window)

• kafka可视化UI：

kafka-Manager github项目地址 (opens new window)

kafka-Manager操作手册 (opens new window)发出消息持久化机制参数****发出消息持久化机制参数