Kafka凭什么能扛下百万级并发?从源码拆解5大高性能核心密码

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19

// Kafka中零拷贝发送的核心代码publiclong transferTo(long position,long count,WritableByteChannel target)throwsIOException{if(target instanceofFileChannel){return transferToFileChannel(position, count,(FileChannel) target);}else{// 传统拷贝路径ByteBuffer buffer =ByteBuffer.allocate((int)Math.min(count,8192));long transferred =0;while(transferred < count){int read = read(buffer, position + transferred);if(read ==-1)break; buffer.flip(); target.write(buffer); transferred += read; buffer.clear();}return transferred;}}

• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53

// 生产者批量消息的核心逻辑publicRecordAppendResult append(TopicPartition tp,long timestamp,byte[] key,byte[] value,Callback callback,long maxTimeToBlock)throwsInterruptedException{// 尝试追加到现有批量Deque<ProducerBatch> dq = getOrCreateDeque(tp);synchronized(dq){if(closed)thrownewIllegalStateException("Cannot send after the producer is closed.");RecordAppendResult appendResult = tryAppend(timestamp, key, value, callback, dq);if(appendResult !=null)return appendResult;}// 创建新的批量int size =Math.max(this.batchSize,Records.LOG_OVERHEAD +Record.recordSize(key, value)); log.trace("Allocating a new {} byte batch for topic {} partition {}", size, tp.topic(), tp.partition());ProducerBatch batch =newProducerBatch(tp, time.milliseconds(), size,this.compressionType,this.keySerializer,this.valueSerializer,this.headers);    FutureRecordMetadata future = null;    try {        future = batch.tryAppend(timestamp, key, value, callback, time.milliseconds());        if (future == null)            throw new IllegalStateException("Should not reach here.");    } catch (Exception e) {        batch.close();        throw e;    }    synchronized (dq) {        if (closed)            throw new IllegalStateException("Cannot send after the producer is closed.");        dq.addLast(batch);        incomplete.add(batch);    }    return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), batch);}

• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78

// Kafka分区的核心管理逻辑public class Partition {    private final String topic;    private final int partition;    private final Log log;    private final LeaderAndIsr leaderAndIsr;    private final List<Replica> replicas;    private final Replica leader;    private final List<Replica> inSyncReplicas;    // 处理消息追加请求    public AppendInfo appendRecords(AppendRecordsRequest request, long now) throws KafkaStorageException {        // 验证leader身份        if (!isLeader)            throw new NotLeaderForPartitionException("Not the leader for partition " + topicPartition);        // 验证ISR副本        if (!isInSync(request.replicaId))            throw new ReplicaNotAvailableException("Replica " + request.replicaId + " is not in ISR for partition " + topicPartition);        // 追加消息到日志        LogAppendInfo logAppendInfo = log.appendAsLeader(request.records, request.isFromClient, now);        // 更新高水位        maybeIncrementHighWatermark(now);        return new AppendInfo(logAppendInfo.lastOffset + 1, logAppendInfo.logStartOffset, logAppendInfo.maxTimestamp, logAppendInfo.offsetOfMaxTimestamp);    }}

• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99

// Kafka异步刷盘的核心代码public void flush() throws IOException {    long start = time.milliseconds();    log.flush();    long end = time.milliseconds();    if (end - start > 1000)        log.warn("Flush for {} took {} ms which is larger than the warn threshold of 1000 ms", this, end - start);}// 定时触发刷盘的任务public class FlushStats implements Runnable {    public void run() {        try {            if (time.milliseconds() - log.lastFlushTime() > flushInterval) {                log.flush();            }        } catch (IOException e) {            log.error("Error flushing log", e);        }    }}

• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107
• 108
• 109
• 110
• 111
• 112
• 113
• 114
• 115
• 116
• 117
• 118
• 119
• 120
• 121
• 122
• 123
• 124

// Kafka线程模型的核心配置public KafkaServer start() throws Throwable {    try {        log.info("starting Kafka server");        // 启动网络线程池        this.networkServer = new SocketServer(config, metrics, time, credentialProvider);        networkServer.startup();        // 启动请求处理线程池        this.kafkaApis = new KafkaApis(networkServer.requestChannel, replicaManager, adminManager,                                        groupCoordinator, transactionCoordinator, config.brokerId,                                        config, metadataCache, metrics, authorizer, time, brokerTopicStats,                                        quotaManagers, clusterId);        kafkaApis.start();        // 启动定时任务线程池        this.scheduler = new KafkaScheduler(config.backgroundThreads);        scheduler.startup();        // 启动其他组件...        log.info("Kafka server started");        return this;    } catch (Throwable t) {        log.error("Fatal error during KafkaServer startup. Prepare to shutdown", t);        shutdown();        throw t;    }}