WARN [ReplicaManager brokerId=1] Produce request with correlation id 12345 from client client1 on partition topic1-15 failed due to request timeout  
ERROR [KafkaRequestHandlerPool-0] Error when handling request: clientId=client2, correlationId=23456, api=PRODUCE  
java.lang.OutOfMemoryError: Java heap space  
WARN [ReplicaFetcherThread-0-3] Error in fetch from broker 3 for partition topic2-5

# 修改Kafka啟動(dòng)腳本中的KAFKA_HEAP_OPTS  
export KAFKA_HEAP_OPTS="-Xms16G -Xmx16G"

# 添加以下GC參數(shù)  
export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent"

# 增加網(wǎng)絡(luò)線程數(shù)，從默認(rèn)的3增加到16  
num.network.threads=16  


# 增加I/O線程數(shù)，從默認(rèn)的8增加到32  
num.io.threads=32  


# 增加請求隊(duì)列大小  
queued.max.requests=1000

# 增加socket接收緩沖區(qū)  
socket.receive.buffer.bytes=1048576  


# 增加socket發(fā)送緩沖區(qū)  
socket.send.buffer.bytes=1048576  


# 增加最大請求大小  
max.request.size=10485760

# 增加批處理大小  
batch.size=131072  


# 設(shè)置linger.ms使消息有時(shí)間進(jìn)行批處理  
linger.ms=50  


# 啟用壓縮  
compression.type=lz4  


# 增加生產(chǎn)者緩沖區(qū)  
buffer.memory=536870912  


# 增加重試次數(shù)和重試間隔  
retries=10  
retry.backoff.ms=100

# 增加單次拉取大小  
fetch.max.bytes=52428800  


# 增加最大拉取等待時(shí)間  
fetch.max.wait.ms=500  


# 增加消費(fèi)者緩沖區(qū)大小  
max.partition.fetch.bytes=10485760

// 在生產(chǎn)者客戶端實(shí)施限流  
Properties props = new Properties();  
props.put("throttle.rate", "10000"); // 每秒限制10000條消息

# 使用kafka-topics.sh工具增加分區(qū)數(shù)  
bin/kafka-topics.sh --bootstrap-server broker1:9092 --alter --topic hot-topic --partitions 60

# 生成分區(qū)重分配計(jì)劃  
bin/kafka-reassign-partitions.sh --bootstrap-server broker1:9092 --generate --topics-to-move-json-file topics.json --broker-list "1,2,3,4,5"  


# 執(zhí)行分區(qū)重分配  
bin/kafka-reassign-partitions.sh --bootstrap-server broker1:9092 --execute --reassignment-json-file reassignment.json

# 增加日志段大小，減少小文件數(shù)量  
log.segment.bytes=1073741824  


# 優(yōu)化日志刷盤策略，避免頻繁刷盤  
log.flush.interval.messages=50000  
log.flush.interval.ms=10000  


# 優(yōu)化日志清理線程數(shù)  
log.cleaner.threads=4

# 增加副本拉取線程數(shù)  
num.replica.fetchers=8  


# 增加副本拉取大小  
replica.fetch.max.bytes=10485760  


# 調(diào)整ISR收縮時(shí)間，避免頻繁的ISR變化  
replica.lag.time.max.ms=30000

# 使用kafka-configs.sh工具動(dòng)態(tài)調(diào)整配置  
bin/kafka-configs.sh --bootstrap-server broker1:9092 --entity-type brokers --entity-name 1 --alter --add-config "num.io.threads=32,num.network.threads=16"

# 配置新的broker節(jié)點(diǎn)  
# server.properties for new brokers  
broker.id=6  
listeners=PLAINTEXT://new-broker:9092  
...  


# 啟動(dòng)新節(jié)點(diǎn)  
bin/kafka-server-start.sh config/server.properties

# 生成遷移計(jì)劃，將部分分區(qū)遷移到新節(jié)點(diǎn)  
bin/kafka-reassign-partitions.sh --bootstrap-server broker1:9092 --generate --topics-to-move-json-file hot-topics.json --broker-list "1,2,3,4,5,6"  


# 執(zhí)行遷移計(jì)劃  
bin/kafka-reassign-partitions.sh --bootstrap-server broker1:9092 --execute --reassignment-json-file migration.json

# 配置多個(gè)日志目錄，分散I/O壓力  
log.dirs=/data1/kafka-logs,/data2/kafka-logs,/data3/kafka-logs,/data4/kafka-logs

# 為高吞吐量主題配置專用存儲(chǔ)策略  
bin/kafka-configs.sh --bootstrap-server broker1:9092 --entity-type topics --entity-name high-throughput-topic --alter --add-config "retention.ms=86400000,cleanup.policy=delete"  


# 為需要長期保存的主題配置壓縮策略  
bin/kafka-configs.sh --bootstrap-server broker1:9092 --entity-type topics --entity-name archive-topic --alter --add-config "cleanup.policy=compact"

// 實(shí)現(xiàn)異步批量發(fā)送模式  
public class OptimizedProducer {  
    private final KafkaProducer<String, String> producer;  
    private final ScheduledExecutorService scheduler;  
    private final ConcurrentLinkedQueue<ProducerRecord<String, String>> messageQueue;  
    private final int batchSize;  


    public OptimizedProducer(Properties props, int batchSize) {  
        this.producer = new KafkaProducer<>(props);  
        this.scheduler = Executors.newScheduledThreadPool(1);  
        this.messageQueue = new ConcurrentLinkedQueue<>();  
        this.batchSize = batchSize;  


        // 定時(shí)批量發(fā)送  
        this.scheduler.scheduleAtFixedRate(this::sendBatch, 100, 100, TimeUnit.MILLISECONDS);  
    }  


    public void send(String topic, String key, String value) {  
        messageQueue.add(new ProducerRecord<>(topic, key, value));  
        if (messageQueue.size() >= batchSize) {  
            sendBatch();  
        }  
    }  


    private void sendBatch() {  
        List<ProducerRecord<String, String>> batch = new ArrayList<>();  
        ProducerRecord<String, String> record;  
        while ((record = messageQueue.poll()) != null && batch.size() < batchSize) {  
            batch.add(record);  
        }  


        for (ProducerRecord<String, String> r : batch) {  
            producer.send(r, (metadata, exception) -> {  
                if (exception != null) {  
                    // 處理失敗，重新入隊(duì)或記錄日志  
                    messageQueue.add(r);  
                }  
            });  
        }  
        producer.flush();  
    }  
}

// 實(shí)現(xiàn)多線程消費(fèi)處理模型  
public class ParallelConsumer {  
    private final Consumer<String, String> consumer;  
    private final ExecutorService executor;  
    private final int numWorkers;  


    public ParallelConsumer(Properties props, int numWorkers) {  
        this.consumer = new KafkaConsumer<>(props);  
        this.numWorkers = numWorkers;  
        this.executor = Executors.newFixedThreadPool(numWorkers);  
    }  


    public void start(String topic, MessageProcessor processor) {  
        consumer.subscribe(Collections.singleton(topic));  


        while (true) {  
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));  
            if (!records.isEmpty()) {  
                // 將記錄分組，并行處理  
                List<Future<?>> futures = new ArrayList<>();  
                for (TopicPartition partition : records.partitions()) {  
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);  
                    futures.add(executor.submit(() -> {  
                        for (ConsumerRecord<String, String> record : partitionRecords) {  
                            processor.process(record);  
                        }  
                    }));  
                }  


                // 等待所有處理完成  
                for (Future<?> future : futures) {  
                    try {  
                        future.get();  
                    } catch (Exception e) {  
                        // 處理異常  
                    }  
                }  


                // 手動(dòng)提交偏移量  
                consumer.commitSync();  
            }  
        }  
    }  


    public interface MessageProcessor {  
        void process(ConsumerRecord<String, String> record);  
    }  
}

# 為高流量主題增加分區(qū)數(shù)  
bin/kafka-topics.sh --bootstrap-server broker1:9092 --alter --topic high-traffic-topic --partitions 100  


# 為關(guān)鍵業(yè)務(wù)主題增加副本因子  
bin/kafka-topics.sh --bootstrap-server broker1:9092 --alter --topic critical-topic --replica-assignment "1:2:3:4:5,2:3:4:5:1,..."

public class BalancedPartitioner implements Partitioner {  
    private Random random = new Random();  


    @Override  
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {  
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);  
        int numPartitions = partitions.size();  


        if (keyBytes == null) {  
            // 隨機(jī)分區(qū)  
            return random.nextInt(numPartitions);  
        } else {  
            // 使用一致性哈希算法，但避免熱點(diǎn)  
            int hashCode = Arrays.hashCode(keyBytes);  
            return Math.abs(hashCode) % numPartitions;  
        }  
    }  


    @Override  
    public void close() {}  


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