Kafka 为什么这么快？(苏三说技术2024年09月18日文章)

分享到：

作者：苏三说技术 | 更新时间：2024-09-20 02:54:25

大家好，我是苏三。Kafka 是一款性能非常优秀的消息队列，每秒处理的消息体量可以达到千万级别。今天来聊一聊 Kafka 高性能背后的技术原理。最近整理了一份10万字的面试宝典，可以免费送给大家，获取方式加我微信：su_san_java，备...

A股板块轮动加剧，跨年大妖来袭，这几只票主力已明显介入！微信搜索关注【研讯小组】公众号（可长按复制），回复666，领取代码！

大家好，我是苏三。

Kafka 是一款性能非常优秀的消息队列，每秒处理的消息体量可以达到千万级别。今天来聊一聊 Kafka 高性能背后的技术原理。

最近整理了一份10万字的面试宝典，可以免费送给大家，获取方式加我微信：su_san_java，备注：面试。

1 批量发送

Kafka 收发消息都是批量进行处理的。我们看一下 Kafka 生产者发送消息的代码：

privateFuturedoSend(ProducerRecordrecord,Callbackcallback){ TopicPartitiontp=null; try{ //省略前面代码 CallbackinterceptCallback=newInterceptorCallback<>(callback,this.interceptors,tp); //把消息追加到之前缓存的这一批消息上 RecordAccumulator.RecordAppendResultresult=accumulator.append(tp,timestamp,serializedKey, serializedValue,headers,interceptCallback,remainingWaitMs); //积累到设置的缓存大小，则发送出去 if(result.batchIsFull||result.newBatchCreated){ log.trace("Wakingupthesendersincetopic{}partition{}iseitherfullorgettinganewbatch",record.topic(),partition); this.sender.wakeup(); } returnresult.future; //handlingexceptionsandrecordtheerrors; //forAPIexceptionsreturntheminthefuture, //forotherexceptionsthrowdirectly }catch/**省略catch代码*/ }

从代码中可以看到，生产者调用 doSend 方法后，并不会直接把消息发送出去，而是把消息缓存起来，缓存消息量达到配置的批量大小后，才会发送出去。

注意：从上面 accumulator.append 代码可以看到，一批消息属于同一个 topic 下面的同一个 partition。

Broker 收到消息后，并不会把批量消息解析成单条消息后落盘，而是作为批量消息进行落盘，同时也会把批量消息直接同步给其他副本。

消费者拉取消息，也不会按照单条进行拉取，而是按照批量进行拉取，拉取到一批消息后，再解析成单条消息进行消费。

使用批量收发消息，减轻了客户端和 Broker 的交互次数，提升了 Broker 处理能力。

2 消息压缩

如果消息体比较大，Kafka 消息吞吐量要达到千万级别，网卡支持的网络传输带宽会是一个瓶颈。Kafka 的解决方案是消息压缩。发送消息时，如果增加参数 compression.type，就可以开启消息压缩：

publicstaticvoidmain(String[]args){ Propertiesprops=newProperties(); props.put("bootstrap.servers","localhost:9092"); props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); //开启消息压缩 props.put("compression.type","gzip"); Producerproducer=newKafkaProducer<>(props); ProducerRecordrecord=newProducerRecord<>("my_topic","key1","value1"); producer.send(record,newCallback(){ @Override publicvoidonCompletion(RecordMetadatametadata,Exceptionexception){ if(exception!=null){ logger.error("sendingmessage error:", e); }else{ logger.info("sendingmessage successful, Offset:", metadata.offset()); } } }); producer.close(); }

如果 compression.type 的值设置为 none，则不开启压缩。那消息是在什么时候进行压缩呢？前面提到过，生产者缓存一批消息后才会发送，在发送这批消息之前就会进行压缩，代码如下：

publicRecordAppendResultappend(TopicPartitiontp, longtimestamp, byte[]key, byte[]value, Header[]headers, Callbackcallback, longmaxTimeToBlock)throwsInterruptedException{ //... try{ //... buffer=free.allocate(size,maxTimeToBlock); synchronized(dq){ //... RecordAppendResultappendResult=tryAppend(timestamp,key,value,headers,callback,dq); if(appendResult!=null){ //Somebodyelsefoundusabatch,returntheonewewaitedfor!Hopefullythisdoesn'thappenoften... returnappendResult; } //这批消息缓存已满，这里进行压缩 MemoryRecordsBuilderrecordsBuilder=recordsBuilder(buffer,maxUsableMagic); ProducerBatchbatch=newProducerBatch(tp,recordsBuilder,time.milliseconds()); FutureRecordMetadatafuture=Utils.notNull(batch.tryAppend(timestamp,key,value,headers,callback,time.milliseconds())); dq.addLast(batch); incomplete.add(batch); //Don'tdeallocatethisbufferinthefinallyblockasit'sbeingusedintherecordbatch buffer=null; returnnewRecordAppendResult(future,dq.size()>1||batch.isFull(),true); } }finally{ if(buffer!=null) free.deallocate(buffer); appendsInProgress.decrementAndGet(); } }

上面的 recordsBuilder 方法最终调用了下面 MemoryRecordsBuilder 的构造方法。

publicMemoryRecordsBuilder(ByteBufferOutputStreambufferStream, bytemagic, CompressionTypecompressionType, TimestampTypetimestampType, longbaseOffset, longlogAppendTime, longproducerId, shortproducerEpoch, intbaseSequence, booleanisTransactional, booleanisControlBatch, intpartitionLeaderEpoch, intwriteLimit){ //省略其他代码 this.appendStream=newDataOutputStream(compressionType.wrapForOutput(this.bufferStream,magic)); }