在需要 ClientId 鉴权的 Kafka 集群中，高效使用 Producer 和 Consumer 的方法_kafka clientid_kafka的consumer clientid

@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP\_SERVERS\_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY\_SERIALIZER\_CLASS\_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        configProps.put(ProducerConfig.VALUE\_SERIALIZER\_CLASS\_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        configProps.put(ProducerConfig.CLIENT\_ID\_CONFIG, "your-client-id");
        return new DefaultKafkaProducerFactory<>(configProps);
    }
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

在需要发送消息的地方，我们注入 KafkaTemplate：

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

void send(String message) {
  ProducerRecord<String, Object> record = new ProducerRecord<>("your-topic-name", message);
  kafkaTemplate.send(record);
}

1
2
3
4
5
6
7
8

然而，当服务启动并尝试发送消息时，发送不成功，出现以下错误：

org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [your-topic-name]

1
2

错误信息提示鉴权失败。通过查看上下文：

INFO [kafka-producer-network-thread | your-client-id-1]

1
2

我们发现 ClientId 由 your-client-id 变成了 your-client-id-1，很明显这是由于第三方包做了手脚。通过查看源码：

发现在创建 Producer 时，会调用 getProducerConfigs 方法，该方法内部会对 ClientId 添加自增的后缀。为了解决这个问题，我们需要重写 DefaultKafkaProducer 的 getProducerConfigs 方法：

public class FixedClientIdKafkaProducerFactory extends DefaultKafkaProducerFactory {
    private final Map<String, Object> configs;
    
    public FixedClientIdKafkaProducerFactory(Map<String, Object> configs) {
        super(configs);
        this.configs = configs;
    }

    @Override
    protected Map<String, Object> getProducerConfigs() {
        final Map<String, Object> newProducerConfigs = new HashMap<>(this.configs);
        checkBootstrap(newProducerConfigs);
        return newProducerConfigs;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

然后在 KafkaProducerConfig 中使用 FixedClientIdKafkaProducerFactory 替换 DefaultKafkaProducerFactory。

消费者

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP\_SERVERS\_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP\_ID\_CONFIG, "your-consumer-group-id");
        props.put(ConsumerConfig.VALUE\_DESERIALIZER\_CLASS\_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ConsumerConfig.KEY\_DESERIALIZER\_CLASS\_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        return new DefaultKafkaConsumerFactory<>(props);
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL\_IMMEDIATE);
        return factory;
    }
}
                  
@Component
public class MyKafkaMessageListener {
    @KafkaListener(topics = "your-topic-name", concurrency = "4", clientIdPrefix = "your-client-id")
    public void listen(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment) {
        // your business logic 
        //……
      
        acknowledgment.acknowledge();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

同样，我们遇到了鉴权失败的问题。错误信息如下：

ERROR[org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1] o.s.k.l.KafkaMessageListenerContainer.error(149): Authentication/Authorization Exception and no authExceptionRetryInterval set
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [your-topic-name]

1
2
3

通过跟踪源码：

我们发现在容器 ConcurrentMessageListenerContainer 的 doStart 方法中，会根据并行度 concurrency 创建多个 KafkaMessageListenerContainer 子容器，然后调用 configureChildContainer 方法配置子容器，并根据 concurrency 和 alwaysClientIdSuffix 参数对 ClientId 添加后缀。如果并行度设置为 1，那么只需要在 KafkaConsumerConfig 中添加如下代码来设置 alwaysClientIdSuffix：

factory.setContainerCustomizer(container -> container.setAlwaysClientIdSuffix(false));

1
2

这样就能确保 ClientId 不会被改变，从而完成鉴权操作。但是为了提高消费能力，我们总归需要设置并行度。因此，根据重写生产者的经验，我们重写 DefaultKafkaConsumerFactory 的 createKafkaConsumer 方法：

public class FixedClientIdKafkaConsumerFactory<K, V> extends DefaultKafkaConsumerFactory<K, V> {
    public FixedClientIdKafkaConsumerFactory(Map configs) {
        super(configs);
    }
    @Override
    protected Consumer<K, V> createKafkaConsumer(String groupId, String clientIdPrefixArg, String clientIdSuffixArg, Properties properties) {
        return super.createKafkaConsumer(groupId, clientIdPrefixArg, null, properties);
    }
}

1
2
3
4
5
6
7
8
9
10

然后在 KafkaConsumerConfig 中使用 FixedClientIdKafkaConsumerFactory 替换 DefaultKafkaConsumerFactory。然而，这次却出现了如下错误：

WARN [main] o.a.k.c.u.AppInfoParser.registerAppInfo(68): Error registering AppInfo mbean
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=your-client-id

1
2
3

在自定义 ConsumerFactory 以确保在多线程环境下共用相同的 ClientId 时，我们必须考虑到启用 JMX 监控时，MBean 的唯一性问题。JMX MBean 是线程级别的，因此如果出现冲突，可能会影响监控功能。

尽管上述日志只是一个 WARN 级别的记录，不会直接影响消费功能，但如果我们希望通过 JMX 实现精准监控，那么必须要解决这个问题。另外，长期看到一系列的 WARN 日志也会令人不安，因此我们决定继续采用自增后缀的 ClientId 策略。

然而，在申请多个 ClientId 时，需要权衡数量。我们需要考虑 Kafka 集群的鉴权能力，同时也要避免后续扩容时频繁申请 ClientId 的问题。

Kafka 的高级 API 确保在同一个消费者组（consumer group）下，每个分区（partition）只能由一个 Consumer 线程消费（1 个 Consumer 线程可以消费多个 partition）。
如上，当一个消费者组包含 5 个消费者并且有 4 个分区时，无论这 5 个消费者是以单实例方式部署还是分布式方式部署，都可能出现某个消费者未被分配到分区的情况，这样会造成线程空跑，占据着资源。为了解决这个问题，我们只需确保总消费者数量小于或等于分区数量。

通常情况下，我们的应用中 Kafka Consumer 仅占据一部分流量。因此，我们是否可以约定单个实例的 Kafka Consumer 的最大并发数为某个固定值呢？比如 6 这样一来，在申请 Topic 时，我们可以一并申请 6 个 ClientId，其命名格式为 -0 至 -5。例如，如果我们申请了一个名为 my-topic-test 的 Topic，那么除了默认生成一个 ClientId 为 my-topic-test-GeMp 之外，我们只需再申请 my-topic-test-GeMp-0、my-topic-test-GeMp-1、my-topic-test-GeMp-2、my-topic-test-GeMp-3、my-topic-test-GeMp-4、my-topic-test-GeMp-5 即可。

在后续需要提升消费能力时，我们可以扩展分区数量。根据分区数量进行横向水平扩容，以保证一个消费者组内的总消费者数量等于分区数量。这样，就不需要再担心 ClientId 的鉴权和 JMX 注册问题了。此外，这也确保了在未来需要提升消费能力并进行分区扩容时，无需再次申请 ClientId。

这种做法可以有效地优化 Kafka Consumer 的管理和扩展，以满足我们的需求。

结论

综上所述，我们总结了在使用 ClientId 进行 Kafka 集群环境下的身份验证时，Kafka 生产者和消费者的一种高效使用方式。

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）

一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

源码讲义、实战项目、讲解视频，并且后续会持续更新**

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-BHXGJx9w-1713017434282)]

一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！