如何为spring-kafka创建的所有MessageListenerContainers设置RecordInterceptor

为spring-kafka创建的所有MessageListenerContainers设置RecordInterceptor，可以通过以下步骤实现：

创建一个自定义的RecordInterceptor类，实现org.apache.kafka.clients.consumer.ConsumerInterceptor接口。该接口包含两个方法：onConsume()和onCommit()。onConsume()方法在消息被消费之前调用，可以在此方法中对消息进行修改或者添加一些额外的处理逻辑。onCommit()方法在消息被提交之前调用，可以在此方法中添加一些额外的提交逻辑。
在自定义的RecordInterceptor类中实现onConsume()方法和onCommit()方法，根据需求对消息进行处理。
在Spring Boot的配置文件中配置Kafka的相关属性，包括bootstrap.servers（Kafka集群的地址）、group.id（消费者组的ID）等。
创建一个自定义的KafkaListenerConfigurer类，实现org.springframework.kafka.annotation.KafkaListenerConfigurer接口。该接口包含一个方法：configureKafkaListeners()。
在自定义的KafkaListenerConfigurer类中实现configureKafkaListeners()方法，在该方法中获取所有的MessageListenerContainers，并为每个MessageListenerContainer设置RecordInterceptor。

以下是一个示例代码：

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.listener.RecordInterceptor;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class CustomKafkaListenerConfigurer implements KafkaListenerConfigurer {

    private final KafkaListenerEndpointRegistry endpointRegistry;

    public CustomKafkaListenerConfigurer(KafkaListenerEndpointRegistry endpointRegistry) {
        this.endpointRegistry = endpointRegistry;
    }

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        Map<String, MessageListenerContainer> containers = endpointRegistry.getListenerContainers();
        for (MessageListenerContainer container : containers.values()) {
            container.setRecordInterceptor(new CustomRecordInterceptor());
        }
    }

    private static class CustomRecordInterceptor<K, V> implements RecordInterceptor<K, V> {

        @Override
        public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
            // 在消息被消费之前的处理逻辑
            // 可以对消息进行修改或者添加额外的处理逻辑
            return records;
        }

        @Override
        public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
            // 在消息被提交之前的处理逻辑
            // 可以添加额外的提交逻辑
        }
    }
}

请注意，上述示例代码中的CustomKafkaListenerConfigurer类需要通过构造函数注入KafkaListenerEndpointRegistry对象，以获取所有的MessageListenerContainers。另外，CustomRecordInterceptor类是一个静态内部类，用于实现RecordInterceptor接口。

这样，为spring-kafka创建的所有MessageListenerContainers就设置了自定义的RecordInterceptor。你可以根据实际需求在CustomRecordInterceptor类中实现相应的逻辑。