Kafka的面试题有哪些

本篇内容主要讲解“Kafka的面试题有哪些”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Kafka的面试题有哪些”吧!

1. RangeAssignor和RoundRobinAssignor的区别，感觉没有区别

答：区别：

按照范围来分配：

对所有消费者按照字典序排序，每个主题中的分区数值序排序。

对主题的各个分区分堆，每个消费者分配一堆。

基于主题的；

一个主题的情况下，将主题的分区分配到消费组的各个消费者：

保证消费者分配的分区数差值最大为1

如果消费组订阅了两个主题，则按照区域，依次对每个主题的分区进行分配：

会导致消费组中不同消费者分配到的分区数差值<font >大于1</font>

轮询的方式列出所有订阅的主题分区，列出所有的消费者线程

对主题分区排序，对消费者排序。

不同消费者分配的分区数，最多多一个，消费者分配的分区数最大差值也就是1而已。

消费组订阅多个分区的时候：

2 生产消息的时候，重试次数到了还失败，缓冲区的那条数据会清掉了吧？是不是要靠客户端捕获异常存下那个消息？

答：是的，需要客户端处理，但是不用捕获异常，直接判断即可。

下面是Sender线程发送消息，处理消息重试主要逻辑的源码展示：

Sender线程：

负责从消息累加器获取消息，并发送：

run方法的实现：

上图中的方法最后，发送消息：

sendProducerData的实现：

上图中的方法最后，调用sendProduceRequests方法发送消息：

上图中sendProduceRequest方法的实现：（注意，此时是单数形式，不是复数形式，也就是说该方法发送单一消息批。）

上图中的方法最后，创建请求对象，发送请求：

callback的handleProduceResponse方法实现：

completeBatch方法实现：

如果消息批在发送的时候出现错误，则先判断是否可以重试，如果可以重试，就将消息批重新入列，等待重新发送：

上图中的canRetry实现：

判断当前消息批是否可以重试，如果可以重试，就需要重试：

将消息批重新入列：

将需要重试的消息批重新入列以待重试：

completeBatch方法中，如果重试次数耗尽，则执行以下代码：

)

failBatch的实现：

上图中调用的方法：

上图中的方法中最后，从累加器中释放该失败的消息批。

问题在于：如果我们发送的消息，重试次数耗尽了，我如何知道是哪个消息发送失败？

如果我想将该失败的消息放到其他的地方，比如进行手动补偿，如何获取该失败的消息？

3. dome1 启动不了，需要配置advertised.listeners 才能启动 是什么原因

配置虚拟机两个IP地址：

kafka的server.properties的配置：

默认配置：

使用PLAINTEXT://:9092，在所有网络接口发布服务。

zookeeper的信息：

默认配置下，106和116两个IP地址都可以使用：

public class MyProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> configs = new HashMap<>();
//        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.100.106:9092");
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.100.116:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);

        RecordMetadata metadata = producer.send(
                new ProducerRecord<String, String>("tp_demo_01", "hello lagou 106-1")
        ).get();

        System.out.println(metadata.topic() + "\t" + metadata.partition() + "\t" + metadata.offset());

        producer.close();

    }
}

如果服务器有两个IP地址，一个是内网，另一个是外网，如何配置，让broker之间通信走内网IP地址，而将外网地址供外网客户端使用？

首先需要配置两个listener：

listeners=PLAINTEXT://192.168.100.106:9092,PLAINTEXT://192.168.100.116:9093
advertised.listeners=PLAINTEXT://192.168.100.116:9093

报错：因为listeners中的PLAINTEXT是监听器的名字。

如果两个监听器，它们的名字不能一样，所以改进配置：

listeners=PLAINTEXT://192.168.100.106:9092,MYCLIENT://192.168.100.116:9093
advertised.listeners=MYCLIENT://192.168.100.116:9093

报错：因为没有给MYCLIENT名字的监听器配置安全协议。

listeners=PLAINTEXT://192.168.100.106:9092,MYCLIENT://192.168.100.116:9093
advertised.listeners=MYCLIENT://192.168.100.116:9093
listener.security.protocol.map=MYCLIENT:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=MYCLIENT

zk结果：

此时使用代码访问：

192.168.100.116:9093可以访问，192.168.100.106:9092不能访问。

则192.168.100.116:9093暴露给客户端使用了。

终极配置：

listeners=PLAINTEXT://192.168.100.106:9092,MYCLIENT://192.168.100.116:9093
advertised.listeners=PLAINTEXT://192.168.100.106:9092,MYCLIENT://192.168.100.116:9093
listener.security.protocol.map=MYCLIENT:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT

1. 一个服务器上有两个IP地址，一个是内网地址，另一个是外网地址。

2. advertised.listeners用于指定要发布到zookeeper上供客户端使用的监听器。

3. 使用listeners让kafka启动的时候，使用指定的协议监听指定的IP地址和端口。

4. 同时指定broker之间通信使用的监听器名称，该监听器使用内网IP地址。

5. advertised.listeners中的另一个供客户端使用。

到此，相信大家对“Kafka的面试题有哪些”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！