SpringBoot 集成 Kafka 配置

时间：2025-01-26 07:08:50

原生模式

        <dependency>
            <groupId></groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.0.0</version>
        </dependency>

自定义分区器

/**
 * 自定义分区器
 *
 * @Author: chen yang
 * @Date: 2023/5/7 11:34
 */
public class CustomerPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        List<PartitionInfo> partitionInfos = (topic);
        int partition = 2;
        return () % partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}

生产者

/**
 * @Author: chen yang
 * @Date: 2023/5/6 21:59
 */
public class Producer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties properties = new Properties();
        // 连接集群
        (ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 序列化类型
        (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ());
        (ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ());

        // 添加自定义分区器
        (ProducerConfig.PARTITIONER_CLASS_CONFIG, ());


        // ：批次大小，默认 16K
        (ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // ：等待时间，默认 0
        (ProducerConfig.LINGER_MS_CONFIG, 1);
        // RecordAccumulator：缓冲区大小，默认 32M：
        (ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // ：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
        (ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");


        // 设置 ack
        // 0:不需要等待数据落盘应答，1:需要等 leader 落盘应答，-1(all):需要等 leader 和所有的 follower(isr队列) 落盘应答
        // type: String, valid values [all, -1, 0, 1], default: all
        (ProducerConfig.ACKS_CONFIG, "all");

        // 设置重试次数，默认为 int 最大值
        (ProducerConfig.RETRIES_CONFIG, 3);

        // 设置事务id
        (ProducerConfig.TRANSACTIONAL_ID_CONFIG, "default_transactional_id_23");

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

        // 开启事务
        ();
        ();

        try {
            // 异步发送
            (new ProducerRecord<>("test", "this is async message"));

            (new ProducerRecord<>("test", "this is a async rollback message!"), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    // 发送消息失败会自动重试，不需要在回调函数中手动重试
                    if ((e)){
                        ("result: " + () + ", partitions: " + ());
                    }
                }
            });

            // 同步发送，只需要在异步发送的基础上再调用 get() 犯法即可
            (new ProducerRecord<>("test", 1,"","this is sync message")).get();
            ();

        }catch (Exception e){
            ();
        }finally {
            ();
        }
    }
}

消费者

/**
 * @Author: chen yang
 * @Date: 2023/7/9 10:37
 */
public class Consumer {

    public static void main(String[] args) {

        Properties properties = new Properties();
        (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        (ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        (ConsumerConfig.GROUP_ID_CONFIG, "consumer_01");
        (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ());
        (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ());

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
              // 订阅 topic
//            ArrayList<String> topics = new ArrayList<>();
//            (topics);

            // 订阅 topic 下的 partition
            ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
            (new TopicPartition("night_topic", 1));
            (topicPartitions);

            // 从指定的 offset 开始消费
//            (new TopicPartition("night_topic", 1), 3);

            ConsumerRecords<String, String> consumerRecords = ((2));
            (::println);

            // 手动提交 offset
            ();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

KafkaTemplate

        <dependency>
            <groupId></groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

配置文件

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      batch-size: 16384
      acks: -1
      retries: 10
      transaction-id-prefix: transaction_05
      buffer-memory: 33554432
      key-serializer: 
      value-serializer: 
      properties:
        linger:
          ms: 2000
        partitioner:
          class: 
    consumer:
      group-id: g_01
      enable-auto-commit: false
      auto-offset-reset: latest
      max-poll-records: 500
#      auto-commit-interval: 2000  autoCommit = false
      key-deserializer: 
      value-deserializer: 
      properties:
        session:
          timeout:
            ms: 120000 # 消费会话超时时间（超过这个时间 consumer 没有发送心跳，就会触发 rebalance 操作）
        request:
          timeout:
            ms: 18000

    listener:
      missing-topics-fatal: false # consumer listener topics 不存在时，启动项目就会报错
      type: batch

自定义分区器

/**
 * @Author: chen yang
 * @Date: 2023/7/8 11:02
 */
@Component
public class CustomerPartitionHandler implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        if (().contains("二")){
            return 2;
        }else if (().contains("一")){
            return 1;
        }else {
            return 0;
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}

生产者

@RestController
@RequiredArgsConstructor
public class HelloController {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send")
    @Transactional // 配置文件中设置了事务id，那么启用时要加上 该注解或者使用 kafka 事务处理
    public Boolean send(String msg){
        for (int i = 0; i < 10; i++) {
            ("", null, "night key - " + i, msg + " - " + i).addCallback(success -> {
                // 成功回调
                if (success == null || () == null){
                    return;
                }
                String topic = ().topic();
                int partition = ().partition();
                long offset = ().offset();
                String key = ().key();
                ("send topic:" + topic +", partition: " + partition + ", key:" + key + ", offset: " + offset);
            }, failure -> {
                // 失败回调
                ("发送消息失败：" + ());
            });
        }
        return true;
    }
}

消费者配置

消费数据过滤
/**
 * 消费数据过滤
 *
 * @Author: chen yang
 * @Date: 2023/7/8 12:20
 */
@Component
public class ConsumerFilterStrategy implements RecordFilterStrategy<String, String> {
    @Override
    public boolean filter(ConsumerRecord<String, String> consumerRecord) {
        // return true: 丢弃消息
        return ().contains("无效数据");
    }
}

消费异常处理类
/**
 * 消费异常处理类
 *
 * @Author: chen yang
 * @Date: 2023/7/8 11:55
 */
@Component
public class ConsumerExceptionHandler implements ConsumerAwareListenerErrorHandler {

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
        ("消费异常：" + () + ", ex: " + ());
        return null;
    }
}

消费者配置 
/**
 * @Author: chen yang
 * @Date: 2023/7/8 12:22
 */
@Configuration
@RequiredArgsConstructor
public class KafkaConsumerConfig {

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final ConsumerFilterStrategy consumerFilterStrategy;


    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> filterContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // @SendTo 使用，console exception:  a KafkaTemplate is required to support replies
        (kafkaTemplate);

        (initConsumerFactory());
        // 设置并发量，小于或等于Topic的分区数,并且要在consumerFactory设置一次拉取的数量
        (1);

        // 设置为批量监听
        (true);

        // 配合RecordFilterStrategy使用，被过滤的信息将被丢弃
        (true);
        (consumerFilterStrategy);
        return factory;
    }



    @Bean
    public ConsumerFactory<String, String> initConsumerFactory(){
        HashMap<String, Object> configs = new HashMap<>();
        (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        (ConsumerConfig.GROUP_ID_CONFIG, "consumer_group_id_02");
        (ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        (ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
        (ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        (ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "");
        (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "");
        return new DefaultKafkaConsumerFactory<>(configs);
    }
}

@KafkaListener

@Component
public class HelloListener {

    @KafkaListener(topicPartitions = {
            @TopicPartition(topic = "", partitions = {"0"})
    }, errorHandler = "consumerExceptionHandler", containerFactory = "filterContainerFactory")
    public void consumer0(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer){
        ("first consumer receive list size: " + ());
        (::println);
        ();
    }


    @KafkaListener(topicPartitions = {
            @TopicPartition(topic = "", partitions = {"1"})
    }, errorHandler = "consumerExceptionHandler")
    @SendTo("singleTopic")
    public String consumer1(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer){
        ("second consumer receive list size: " + ());
        (::println);
        ();
        return "@SendTo annotation msg";
    }


    @KafkaListener(topicPartitions = {
            @TopicPartition(topic = "", partitions = {"2"})
    }, errorHandler = "consumerExceptionHandler")
    public void consumer2(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer){
        ("third consumer receive list size: " + ());
        (::println);
        ();
    }
}

相关文章