Kafka生产者高吞吐量优化实践：突破百万消息/秒的瓶颈

spring.kafka.bootstrap-servers=PLAINTEXT://localhost:9092
# 生产者配置
spring.kafka.producer.retries=0 # 禁用重试，避免延迟
spring.kafka.producer.batch-size=163840 # 增加批次大小，例如160KB
spring.kafka.producer.linger-ms=100 # 增加等待时间，形成更大批次
spring.kafka.producer.acks=0 # 最高吞吐量模式
spring.kafka.producer.enable-idempotence=false # 禁用幂等性，配合acks=0
spring.kafka.producer.compression-type=lz4 # 启用压缩，减少网络带宽
spring.kafka.producer.buffer-memory=33554432 # 生产者发送缓冲区大小，默认32MB，可根据内存适当调整

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaAdmin;
import org.apache.kafka.clients.admin.AdminClientConfig;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaTopicConfig {

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return new KafkaAdmin(configs);
    }

    @Bean
    public NewTopic testTopic() {
        // 创建一个名为"test-topic"的主题，6个分区，1个副本
        // 注意：这里将min.insync.replicas设置为1
        Map<String, String> topicConfigs = new HashMap<>();
        topicConfigs.put("min.insync.replicas", "1"); // 关键配置

        return new NewTopic("test-topic", 6, (short) 1).configs(topicConfigs);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

@Component
public class KafkaProducerJob {

    private static final String TOPIC = "test-topic";
    private static final int MESSAGES_PER_RUN = 1_000_000; // 目标消息数
    private static final int NUM_THREADS = 8; // 根据CPU核心数和Kafka分区数调整
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);
    private final AtomicLong totalMessagesSent = new AtomicLong(0);

    @Autowired
    public KafkaProducerJob(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // 可以根据需要调整调度策略
    @Scheduled(fixedDelay = 15000)
    public void scheduleProducerTask() {
        System.out.println("--- Starting Kafka Producer Task ---");
        long startTime = System.currentTimeMillis();
        totalMessagesSent.set(0); // 重置计数器

        for (int i = 0; i < NUM_THREADS; i++) {
            executorService.submit(() -> produceMessages(MESSAGES_PER_RUN / NUM_THREADS));
        }

        // 等待所有任务完成
        executorService.shutdown();
        try {
            // 最多等待5分钟，确保任务完成
            if (!executorService.awaitTermination(5, TimeUnit.MINUTES)) {
                System.err.println("Producer tasks did not terminate in time.");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Producer task interrupted: " + e.getMessage());
        } finally {
            // 重新初始化线程池以便下次调度使用
            // 注意：在实际生产中，可能更倾向于一次性启动或使用更复杂的线程池管理
            // 这里为了演示调度，暂时这样处理，但更好的做法是使用@Async方法配合ThreadPoolTaskExecutor
            // executorService = Executors.newFixedThreadPool(NUM_THREADS);
        }


        long endTime = System.currentTimeMillis();
        long duration = endTime - startTime;
        long actualSent = totalMessagesSent.get();
        double messagesPerSecond = (double) actualSent / (duration / 1000.0);

        System.out.printf("--- Finished Kafka Producer Task ---%n");
        System.out.printf("Total messages sent: %d%n", actualSent);
        System.out.printf("Time taken: %d ms%n", duration);
        System.out.printf("Throughput: %.2f messages/second%n", messagesPerSecond);
    }

    private void produceMessages(int count) {
        for (int i = 0; i < count; i++) {
            try {
                String message = "Test Message sadg sad-" + totalMessagesSent.incrementAndGet();
                kafkaTemplate.send(TOPIC, message);
            } catch (Exception e) {
                System.err.println("Error sending message: " + e.getMessage());
                // 生产环境中应有更完善的错误处理和日志记录
            }
        }
    }
}

# 假设Kafka安装在/opt/kafka目录下
# 运行生产者性能测试
/opt/kafka/bin/kafka-producer-perf-test.sh \
    --topic test-topic \
    --num-records 10000000 \
    --record-size 100 \
    --throughput -1 \
    --producer-props \
        bootstrap.servers=localhost:9092 \
        acks=0 \
        linger.ms=100 \
        batch.size=163840 \
        compression.type=lz4 \
        buffer.memory=33554432