kafka几道面试题

Topic 副本因子个数：replication.factor >= 3
同步副本列表(ISR)：min.insync.replicas = 2
禁用unclean选举：unclean.leader.election.enable=false

同步方式下配置：
producer.type=sync
request.required.acks=1
副本数量>=2
增加重试次数

producer.type=async 
request.required.acks=1 
queue.buffering.max.ms=5000 
queue.buffering.max.messages=10000 
queue.enqueue.timeout.ms = -1 
batch.num.messages=200
queue.buffering.max.ms=5000
通过buffer来进行控制数据的发送，有两个值来进行控制，缓冲时间阈值与缓冲消息的数量阈值，
如果buffer满了数据还没有发送出去，如果设置的是立即清理模式，风险很大，一定要设置为阻塞模式

1：关闭自动 offset，手动提交offset
设置 enable.auto.commit = false ， 默认值true，自动提交
使用kafka的Consumer的类，用方法consumer.commitSync()提交
或者使用spring-kafka的 Acknowledgment类，用方法ack.acknowledge()提交（推荐使用）
2：另一个方法同样需要手动commit offset，另外在consumer端再将所有fetch到的数据缓存到queue里，
当把queue里所有的数据处理完之后，再批量提交offset，这样就能保证只有处理完的数据才被commit。

1） Producer exactly-once
enable.idempotence=true  
分区副本数>= 2
isr >=2
ProducerID+SequenceNumber+Ack=-1（幂等性）
2）Consumer exactly-once
手动维护并提交偏移量。
1：设置enable.auto.commit=false，关闭自动提交偏移量
2：借助外部数据库，如redis的pipeline，mysql的事务机制管理存储偏移量
再同一事物中，在消息被处理完之后在提交偏移量。并更新偏移量。
否则消息需回滚，并获取到上一次偏移量的位置从新进行处理。

# cat /proc/sys/net/core/rmem_max 
124928
# cat /proc/sys/net/core/wmem_max 
124928

#vi /etc/sysctl.conf
fs.file-max = 32000
#vi /etc/security/limits.conf
yourusersoftnofile10000
youruserhardnofile30000

import org.apache.kafka.clients.producer.*; 
import java.util.Properties; 
/**
 * 设置核心的参数，提升发送消息的效率
 */
public class ProducerCustomParams {
    public static void main(String[] args) throws Exception {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "IP:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // batch.size：批次大小，默认 16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // linger.ms：等待时间，默认 0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        System.out.println("开始发送数据");
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 100; i++) {
            kafkaProducer.send(new ProducerRecord<>("zcy222", "congge " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e)
 {
                    if (e == null) {
                        System.out.println("发送成功");
                        System.out.println("主题:" + metadata.topic());
                        System.out.println("分区:" + metadata.partition());
                    }
                }
            });
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

1、增加broker节点，增加分区数量，提高并行度
2、修改单个消费为批量消费
3、增加单线程消费为线程池异步消费
4、缩短批次时间间隔；
5、老版本SparkStreaming控制消费的速率（spark.streaming.kafka.maxRatePerPartition），可以控制最大的消费速率，在参数中设置；新版本设置背压机制实现消费处理的动态平衡。
6.对代码进行优化，尽可能的一次性计算多个结果，减少shuffer过程；
7.处理的结果如果过多，可以将数据保存到MySQL集群、MongoDB集群【支持事物】或ES【不支持事物】，增大吞吐量
8、消费线程将拉取的消息放到一个滑动窗口中，通过滑动窗口控制拉取的速度
9、对于倾斜的key加以处理，加随机数扽等方式打散
参考：http://niyanchun.com/kafka-multi-thread-consumer.html

协调者在管理消费者组成员的时候，会对消费者组标定状态。在协调者的视角，消费者组一共有 5 种状态：
● Empty：表示组内没有任何成员，但可能存在已经提交的位移数据，而且这些数据还没有过期。
● Dead：表示组内没有任何成员，而且元信息已经被移除。
● PreparingRebalance：表示准备开始 Rebalance，所有消费成员都需要重新向协调者请求加入消费者组。
● CompletingRebalance：表示所有的成员已经重新加入消费者组，正在等待分配方案。
● Stable：表示稳定状态，也就是完成 Rebalance 后可以正常消费数据的状态。

JoinGroup请求：consumer请求加入组
SyncGroup请求：group leader把分配方案同步更新到组内所有成员中。
Heartbeat请求：consumer定期向coordinator汇报心跳表明自己依然存活。
LeaveGroup请求：consumer主动通知coordinator自己将要离开consumergroup。
DescribeGroup请求：查看组的所有的所有信息，包括成员信息、协议信息、分配方案、以及订阅信息等。该请求主要供管理员使用，coordinator不使用该请求实现rebalance。

num.recovery.threads.per.data.dir
对于如下 3 种情况，Kafka 会使用可配置的线程池来处理日志片段：
    器正常启动，用于打开每个分区的日志片段；
    服务器崩溃后重启，用于检查和截短每个分区的日志片段；
    服务器正常关闭，用于关闭日志片段。
默认情况下，每个日志目录只使用一个线程。因为这些线程只是在服务器启动和关闭时会用到，所以完全可以设置大量的线程来达到并行操作的目的。特别是对于包含大量分区的服务器来说，一旦发生崩溃，在进行恢复时使用并行操作可能会省下数小时的时间。
设置此参数时需要注意，所配置的数字对应的是 log.dirs 指定的单个日志目录。
也就是说，如果 num.recovery.threads.per.data.dir 被设为 8，并且且log.dir 指定了 3 个路径，那么总共需要 24 个线程。
num.partitions
设置合理的分区数，分区数=Tt /min（Tp，Tc），建议设置3~10个分区
log.retention.ms
Kafka 通常根据时间来决定数据可以被保留多久。默认使用 log.retention.hours 参数来配置时间，默认值为 168 小时，也就是一周。除此以外，还有其他两个参数 log.retention.minutes 和 log.retention.ms 。
这 3 个参数的作用是一样的，都是决定消息多久以后会被删除，不过还是推荐使用 log.retention.ms 。
如果指定了不止一个参数，Kafka 会优先使用具有最小值的那个参数。
log.retention.bytes
另一种方式是通过保留的消息字节数来判断消息是否过期。它的值通过参数log.retention.bytes 来指定，作用在每一个分区上。
也就是说，如果有一个包含 8 个分区的主题，并且 log.retention.bytes 被设为 1GB，那么这个主题最多可以保留 8GB 的数据。
所以，当主题的分区个数增加时，整个主题可以保留的数据也随之增加。
根据字节大小和时间保留数据
如果同时指定了 log.retention.bytes 和 log.retention.ms （或者另一个时间参数），只要任意一个条件得到满足，消息就会被删除。
例如，假设 log.retention.ms 设置为 86 400 000（也就是 1 天），log.retention.bytes 设置为 1 000 000 000（也就是 1GB），
如果消息字节总数在不到一天的时间就超过了 1GB，那么多出来的部分就会被删除。相反，如果消息字节总数小于 1GB，那么一天之后这些消息也会被删除，尽管分区的数据总量小于 1GB。
log.segment.bytes
此设置作用在作用在日志片段上。当日志片段大小达到 log.segment.bytes 指定的上限（默认是 1GB）时，当前日志片段就会被关闭，一个新的日志片段被打开。如果一个日志片段被关闭，就开始等待过期。
这个参数的值越小，就会越频繁地关闭和分配新文件，从而降低磁盘写入的整体效率。
log.segment.ms
此设置指定了多长时间之后日志片段会被关闭。日志片段会在大小或时间达到上限时被关闭，就看哪个条件先得到满足。
message.max.bytes
此设置用于设置 message.max.bytes 参数来限制单个消息的大小，默认值是 10000 000，也就是 1MB
num.network.threads
创建 Processor 处理网络请求线程个数，建议设置为 Broker 当前CPU核心数*2，这个值太低经常出现网络空闲太低而缺失副本。
num.io.threads
创建 KafkaRequestHandler 处理具体请求线程个数，建议设置为Broker磁盘个数*2。
num.replica.fetchers
建议设置为CPU核心数/4，适当提高可以提升CPU利用率及 Follower同步 Leader数据当并行度
compression.type
建议采用lz4压缩类型，压缩可以提升CPU利用率同时可以减少网络传输数据量。
log.flush.xxx
log.flush.scheduler.interval.ms
log.flush.interval.ms
log.flush.interval.messages
这几个参数表示日志数据刷新到磁盘的策略，应该保持默认配置，刷盘策略让操作系统去完成，由操作系统来决定什么时候把数据刷盘；如果设置来这个参数，可能对吞吐量影响非常大
auto.leader.rebalance.enable
表示是否开启leader自动负载均衡，默认true；我们应该把这个参数设置为false，因为自动负载均衡不可控，可能影响集群性能和稳定。

ack
kakfa消息语义。
buffer.memory
该参数用来设置生产者内存缓冲区的大小，生产者用它缓冲要发送到服务器的消息。如果应用程序发送消息的速度超过发送到服务器的速度，会导致生产者空间不足。
这个时候， send() 方法调用要么被阻塞，要么抛出异常，取决于如何设置block.on.buffer.full 参数（在 0.9.0.0 版本里被替换成了 max.block.ms ，表示在抛出异常之前可以阻塞一段时间）。
compression.type
默认情况下，消息发送时不会被压缩。该参数可以设置为 snappy 、gzip 或 lz4。
snappy 压缩算法占用较少的 CPU，却能提供较好的性能和相当可观的压缩比，如果比较关注性能和网络带宽，可以使用这种算法。
gzip 压缩算法一般会占用较多的 CPU，但会提供更高的压缩比，所以如果网络带宽比较有限，可以使用这种算法。
使用压缩可以降低网络传输开销和存储开销，而这往往是向 Kafka 发送消息的瓶颈所在。
retries
消息重试机制。默认情况下，生产者会在每次重试之间等待 100ms，不过可以通过 retry.backoff.ms 参数来改变这个时间间隔。
batch.size
批量发送消息大小。该参数指定了一个批次可以使用的内存大小，按照字节数计算（而不是消息个数）。
linger.ms
该参数指定了生产者在发送批次之前等待更多消息加入批次的时间。KafkaProducer 会在批次填满或 linger.ms 达到上限时把批次发送出去。
默认情况下，只要有可用的线程，生产者就会把消息发送出去，就算批次里只有一个消息。
max.in.flight.requests.per.connection
该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高，就会占用越多的内存，不过也会提升吞吐量。
把它设为 1 可以保证消息是按照发送的顺序写入服务器的，即使发生了重试。
timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms
request.timeout.ms 指定了生产者在发送数据时等待服务器返回响应的时间，
metadata.fetch.timeout.ms 指定了生产者在获取元数据（比如目标分区的首领是谁）时等待服务器返回响应的时间。
max.block.ms
该参数指定了在调用 send() 方法或使用 partitionsFor() 方法获取元数据时生产者的阻塞时间。
当生产者的发送缓冲区已满，或者没有可用的元数据时，这些方法就会阻塞。
在阻塞时间达到 max.block.ms 时，生产者会抛出超时异常。
max.request.size
该参数用于控制生产者发送的请求大小。它可以指能发送的单个消息的最大值，也可以指单个请求里所有消息总的大小。
receive.buffer.bytes 和 send.buffer.by
这两个参数分别指定了 TCP socket 接收和发送数据包的缓冲区大小。如果它们被设为 -1，就使用操作系统的默认值。
如果生产者或消费者与 broker 处于不同的数据中心，那么可以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。

fetch.min.bytes
该属性指定了消费者从服务器获取记录的最小字节数。如果没有很多可用数据，但消费者的 CPU 使用率却很高，那么就需要把该属性的值设得比默认值大。
如果消费者的数量比较多，把该属性的值设置得大一点可以降低 broker 的工作负载。
fetch.max.wait.ms
指定 broker 的等待时间，默认是 500ms。如果要降低潜在的延迟（为了满足 SLA），可以把该参数值设置得小一些。
如果 fetch.max.wait.ms 被设为 100ms，并且 fetch.min.bytes 被设为 1MB，那么 Kafka 在收到消费者的请求后，
要么返回 1MB 数据，要么在 100ms 后返回所有可用的数据，就看哪个条件先得到满足。
max.partition.fetch.bytes
该属性指定了服务器从每个分区里返回给消费者的最大字节数。它的默认值是 1MB。在为消费者分配内存时，可以给它们多分配一些，因为如果群组里有消费者发生崩溃，剩下的消费者需要处理更多的分区。
max.partition.fetch.bytes 的值必须比 broker 能够接收的最大消息的字节数（通过 max.message.size 属性配置）大，否则消费者可能无法读取这些消息，导致消费者一直挂起重试。
在设置该属性时，另一个需要考虑的因素是消费者处理数据的时间。消费者需要频繁调用 poll() 方法来避免会话过期和发生分区再均衡，如果单次调用 poll() 返回的数据太多，
消费者需要更多的时间来处理，可能无法及时进行下一个轮询来避免会话过期。如果出现这种情况，可以把 max.partition.fetch.bytes 值改小，或者延长会话过期时间。
session.timeout.ms
该属性指定了消费者在被认为死亡之前可以与服务器断开连接的时间，默认是 3s。
该属性与 heartbeat.interval.ms 紧密相关。heartbeat.interval.ms 指定了 poll() 方法向协调器发送心跳的频率，session.timeout.ms 则指定了消费者可以多久不发送心跳。
所以，一般需要同时修改这两个属性，heartbeat.interval.ms 必须比 session.timeout.ms 小，一般是 session.timeout.ms 的三分之一
auto.offset.reset
该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下（因消费者长时间失效，包含偏移量的记录已经过时并被删除）该作何处理。它的默认值是 latest 
enable.auto.commit
该属性指定了消费者是否自动提交偏移量，默认值是 true 。为了尽量避免出现重复数据和数据丢失，可以把它设为 false ，由自己控制何时提交偏移量。
如果把它设为 true ，还可以通过配置 auto.commit.interval.ms 属性来控制提交的频率。
partition.assignment.strategy
通过设置partition.assignment.strategy 来选择分区策略，有Range，RoundRobin，sticky三种策略。
max.poll.records
该属性用于控制单次调用 call() 方法能够返回的记录数量，可以帮你控制在轮询里需要处理的数据量。
eceive.buffer.bytes 和 send.buffer.bytes
socket 在读写数据时用到的 TCP 缓冲区也可以设置大小。如果它们被设为 -1，就使用操作系统的默认值。如果生产者或消费者与 broker 处于不同的数据中心内，
可以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。

kakfa通过时间轮来处理延迟任务，只将时间轮的槽保存到延迟队列，大大的减少了延迟队列的元素数量，这样对于元素的增加删除性能有很大提高；
kafka通过阻塞的方式poll延迟队列的，减少了大量的空转；
为了保证线程安全，灵活运用读写锁、原子对象、synchronized控制时间轮的操作；