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，可根据内存适当调整

ControlNet

AI图像生成的规则改变者，通过添加额外条件来控制SD模型

81

查看详情

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