介绍Kafka是由 LinkedIn 开发并开源的一个分布式消息流平台现由 Apache 基金会维护。它主要用于实时处理和传输大规模数据流并结合了消息队列和流处理的功能。Kafka 的核心特性包括高吞吐量、低延迟、可扩展性和可靠性。KafkaKafka安装到启动测试步骤安装官网下载地址Downloads | Apache Kafka注意kafka需要依赖zookeeper来协调和管理集群中的元数据信息我这边以Windows系统和kafka_2.13-3.5.1版本演示它自带了zookeeper就无需额外去下载zookeeper了。下载kafka_2.13-3.5.1.tgz压缩包解压下载好的压缩包到没有空格和中文的文件夹中配置修改config目录下的图中这两个配置文件修改server.properties文件然后保存同理修改zookeeper.properties文件然后保存启动启动 kafka 前需要先启动 zookeeper当然你也可以直接winR在左下角输入cmd然后通过cd命令来切换盘符。在小黑窗输入命令bin\windows\zookeeper-server-start.bat config\zookeeper.properties看到如图则说明启动zookeeper成功启动kafka同样的在该目录下打开窗口输入命令bin\windows\kafka-server-start.bat config\server.properties看到如图则说明启动 kafka 成功测试创建topic如果有学过 RabbitMQ消息队列的对topic会更容易理解对于RabbitMQ来说创建一个queue队列其实就是创建一个exchange和queue我们可以向exchange里发送消息由exchange来转发消息到queue。对于Kafka来说创建一个topic其实就是创建一个接收消息的主题主题中有多个分区我们可以向这个主题里发送消息。kafka的主题里边可以包含多个分区我们发送消息到主题如果不指定分区主题会根据一些策略将消息分配到其中一个分区相当于一个主题里有多个list你往 list 里添加数据的时候你就直接通知主题由主题来给你选一个list 添加数据。创建topic命令bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 8 --topic test-topic创建生产者创建生产者命令bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test-topic创建消费者创建消费者命令bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test-topic我们来在生产者输入一些测试消息向消费者发送消息看看消费者是否能接收到Kafka基础部分核心架构和概念一个典型的 Kafka架构 包括若干 Producer、若干Broker、若干Consumer 以及一个ZooKeeper集群。其中ZooKeeper是 Kafka 用来负责集群元数据的管理、控制器的选举等操作的。Producer将消息发送到BrokerBroker负责将收到的消息存储到磁盘中而Consumer负责从Broker订阅并消费消息。对于Kafka而言Broker可以简单地看作一个独立的Kafka服务节点或Kafka服务实例。大多数情况下也可以将Broker看作一台Kafka服务器前提是这台服务器上只部署了一个Kafka实例一个或多个Broker组成一个Kafka集群。在Kafka中还有两个特别重要的概念主题(Topic)与分区(Partition)。主题是一个逻辑上的概念它还可以细分为多个分区一个分区只属于单个主题很多时候也会把分区称为主题分区(TopicPartition)。同一主题下的不同分区包含的消息是不同的分区在存储层面可以看作一个可追加的日志(Log)文件消息在被追加到分区日志文件的时候都会分配一个特定的偏移量offsetrabbitmq里边是消息序号为什么rabbitmq不能用offset呢因为ratbbitmq就是一个queue可kafka就不一样它不是一个queue它压根儿就没有队列的概念人家是靠文件管理的用offet搞的所以更准确的来说kafka是一个消息中间件这也是rabbitmq和kakfa的核心区别。offset是消息在分区中的唯一标识rabbitmq里边是通过消息i做唯一标识的)Kafka通过它来保证消息在分区内的顺序性offset是存储地址的偏移量是能保证顺序的后存储进来的消息肯定是在文件的末尾rabbitmq的消息顺序性-人家rabbitmq就是一个queue天生的先进先出当然了人家也有唯一自增id标识)。不过offset并不跨越分区也就是说Kafka保证的是分区有序而不是主题有序不同分区包含的消息是不同的所以不同的消息属于不同的分区所以不同的消息的offset属于不同的分区所以offset不能再分区共享rabbitmq在同一个queue里是保证消息顺序的在所有queue里不能保证。如下图来举例说明主题分区内容图中的主题包含5个分区那么对于这5个分区来说能不能把这5个分区分配到不同的broker中也就是说假如有3台kafka服务节点a,b,c一个主题中的这5个分区可不可以分散的分配到abc这三个节点中?答案是可以的因为主题就是一个逻辑概念主题就是为了给消息做负载的那么对于往哪台服务器上的主题分区分配都ok。可为什么要做成这样呢为了扩展性想象下如果一个主题的所有分区都只能分配在一个kafka服务节点上我们如何扩展呢?一个kafka的服务节点资源是有限的。(类似比如springcloud gateway(kafka)就可看成一个主题转发请求到服务节点(分区)。第3个分区后边还有3个副本我们称之为分区副本说明第3个分区有3个副本保证数据的可靠性。分区副本也是存储到不同的kafka节点的而且kafka的自身设计会优先选择将副本分到不同的kafka服务节点。副本还有一些内容副本首领--副本首领选举--acks生产者参数都和副本有关。副本首领和副本首领选举一个分区可以有多个副本但这些副本里只有一个是首领Leader剩下的都是跟随者。生产者发消息和消费者拉消息都只跟首领打交道跟随者不对外服务它唯一的任务就是从首领那里同步数据保证自己跟首领的数据一致。那首领挂了怎么办这就涉及到首领选举。Kafka会从跟随者里选一个出来当新首领。但不是所有跟随者都有资格——只有跟得上节奏的跟随者才有选举权。Kafka用了一个叫ISRIn-Sync Replicas同步副本集合的机制来管理这个。一个跟随者要是长时间没跟首领同步、数据落后太多就会被踢出ISR失去选举资格。默认情况下如果10秒replica.lag.time.max.ms没同步上就会被踢出去。这样做是为了保证选出来的新首领一定是数据完整的不会丢消息。谁来负责选举呢Kafka集群里有一个特殊的Broker叫Controller专门管这件事。它监听所有Broker的上下线一旦发现有Broker挂了它就从ISR里选一个新首领。同一个时刻只有一个ControllerController自己挂了也会重新选一个。拦截器及自定义拦截器是生产者端的一个扩展机制允许你在消息发送前后插入自定义的逻辑。它有两个钩子onSend在消息发送之前、序列化之前被调用。你可以在这里修改消息内容、添加header、记录日志甚至可以返回null来丢弃这条消息。onAcknowledgement在收到Broker的确认之后被调用不管发送成功还是失败都会触发。你可以在这里统计成功率、记录延迟。close生产者关闭的时候调用可以用来输出最终统计之类的收尾工作。多个拦截器可以串起来组成一个链按配置顺序依次执行。自定义拦截器需要实现ProducerInterceptor接口。举一个实际的例子你现在想给每条消息自动加一个发送时间戳同时统计整个生产者的发送成功率可以这样写public class MyInterceptor implements ProducerInterceptorString, String { private long successCount 0; private long failCount 0; Override public ProducerRecordString, String onSend(ProducerRecordString, String record) { // 给header里塞一个发送时间戳 record.headers().add(send-time, String.valueOf(System.currentTimeMillis()).getBytes()); return record; } Override public void onAcknowledgement(RecordMetadata metadata, Exception e) { if (e null) { successCount; } else { failCount; } } Override public void close() { // 生产者关闭时打出统计 System.out.println(成功: successCount , 失败: failCount); } Override public void configure(MapString, ? configs) {} }配置的时候在ProducerConfig.INTERCEPTOR_CLASSES_CONFIG里填上类的全路径多个拦截器用逗号隔开就行了。拦截器常见的用途包括消息审计记录每条消息的trace信息、自动加traceId、消息过滤return null、简单的限流计数等等。序列化及自定义序列化就是把你的消息对象转成byte数组因为网络传输只认字节流。消费者收到之后再做反序列化把byte数组还原成消息对象。Kafka内置了几种基本类型的序列化器StringSerializer、IntegerSerializer、LongSerializer、DoubleSerializer、ByteArraySerializer、ByteBufferSerializer。大多数时候我们发JSON字符串直接用StringSerializer就够了。但如果你的消息比较复杂发的是一个User对象之类的那就需要自定义序列化器。实现Serializer接口就行核心方法是serialize(String topic, T data)返回byte[]。自定义一个JSON序列化器很简单拿Jackson来做public class JsonSerializerT implements SerializerT { private final ObjectMapper mapper new ObjectMapper(); Override public byte[] serialize(String topic, T data) { if (data null) return null; try { return mapper.writeValueAsBytes(data); } catch (Exception e) { throw new SerializationException(序列化失败, e); } } }反序列化器同理实现Deserializer接口。不管你用什么序列化方式JSON、Protobuf、Avro消费者端的反序列化器必须跟你生产者端的序列化器配对不然收到一堆字节解不出来。这里有一个很重要的点序列化方案一旦上线就很难改。因为所有消费者都要跟着变。所以选型要慎重一开始就定好。内部微服务间用JSON最简单省事高吞吐场景可以考虑Protobuf大数据生态可以考虑Avro配合Schema Registry。分区器及自定义分区器的作用就是决定消息发到主题的哪个分区。消息到达生产者后经过拦截器、序列化器最后到分区器这里来决定分区号。默认的分区器DefaultPartitioner的逻辑是这样的首先如果你在发送消息的时候明确指定了partition字段直接用你指定的分区。其次如果消息有key且不为null就对key进行hash然后用hash值对分区数取模这样同一个key永远去同一个分区——保证了按key有序。最后如果key为null就用Sticky Partitioning粘性分区后面专门讲。自定义分区器需要实现Partitioner接口核心方法是partition。举个例子你的消息key是地区名华东、华南、华北你想让同一个地区的消息去同一个分区保证同一个地区的消息有序消费public class RegionPartitioner implements Partitioner { Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { int partitionCount cluster.partitionsForTopic(topic).size(); if (keyBytes null) { return ThreadLocalRandom.current().nextInt(partitionCount); } // key是地区名hash取模保证同一个地区去同一个分区 return Math.abs(key.toString().hashCode()) % partitionCount; } Override public void close() {} Override public void configure(MapString, ? configs) {} }配置的时候在ProducerConfig.PARTITIONER_CLASS_CONFIG里填上自定义分区器的全路径即可。默认分区器对null key的处理当key为null的时候Kafka 2.4之后的默认分区器用的不是传统的Round-Robin轮询而是Sticky Partitioning粘性分区。传统的轮询是发一条就换一个分区每条消息都可能去不同的分区msg1 → partition 0msg2 → partition 1msg3 → partition 2msg4 → partition 0粘性分区不一样它不是马上换分区而是在同一个分区上把消息攒够一批batch满了再换分区msg1 → partition 0msg2 → partition 0 ← 粘在partition 0msg3 → partition 0 ← 继续粘msg4 → partition 1 ← 这个batch满了换partition 1继续粘为什么改成粘性分区为了减少网络开销。Kafka的消息是按批次发送的——多条消息攒成一个batch然后一次性发一个网络请求。如果每条消息都换分区那每个batch就凑不了几条消息可能每次就发一条网络利用率极低。粘性分区让同一个batch全发往同一个分区batch填满了才换分区。这样既保持了负载均衡长时间看每个分区的消息量是均匀的又提高了批次发送的效率。说白了就是鱼和熊掌都想要Round-Robin的均匀性加上批次发送的高性能。异步发送消息的callback机制Kafka生产者发消息的时候send()方法是异步的它返回一个FutureRecordMetadata。你如果想拿到发送结果有两种方式方式一同步等。调用future.get()这行代码会阻塞一直等到Broker返回确认。简单粗暴但性能差发一条等一条吞吐量上不去。方式二异步回调。send()方法可以传一个Callback参数重写onCompletion方法。发送成功后Broker返回确认你的onCompletion方法会被调用在里面可以拿到RecordMetadata包含topic、partition、offset这些信息也可以捕获到发送失败时的异常。这样不阻塞主线程发完就继续发下一条性能好很多。实际项目里基本都用异步回调模板大概是这样的producer.send(record, (metadata, exception) - { if (exception ! null) { // 发送失败记日志加监控告警 log.error(发送失败: topic{}, error{}, record.topic(), exception.getMessage()); } else { // 发送成功可以用metadata.topic()/partition()/offset() log.info(发送成功: offset{}, metadata.offset()); } });有几个细节要注意第一Callback是在生产者的IO线程里执行的不是你自己开的线程。所以回调逻辑必须轻量快速不要在Callback里做耗时操作比如数据库写入、远程调用这些否则会拖慢整个生产者的发送速度。第二Callback的执行顺序不是按发送顺序来的而是按Broker确认到达的顺序。如果你先后发了msgA和msgB但msgA去了partition 0msgB去了partition 1两个分区的确认时间不一样有可能msgB的callback先执行、msgA的callback后执行。这一点要注意别依赖callback顺序来判断消息发送顺序。第三Callback里一定要try-catch住所有异常。Callback里如果抛出一个未捕获的异常会导致生产者IO线程崩掉后续所有消息都发不出去——这是一个很常见的坑。生产者acks参数acks参数控制的是生产者在什么条件下才算消息发送成功。它跟副本机制有直接关系因为acks决定了生产者要等几个副本确认。acks0生产者发完消息就不管了不等任何确认。最快但也最不可靠消息丢了都不知道。适合那种丢几条也无所谓的场景比如日志采集。acks1生产者等首领写入成功就认为发送成功了不等等跟随者同步。如果首领写完就挂了、跟随者还没来得及同步这条消息就丢了。这是默认值适合一般业务大多数场景够用。acksall或者-1生产者要等所有ISR里的副本都写入成功才认为发送成功。最可靠但性能也最低。适合订单、支付这种不能丢消息的场景。这里还有一个参数要配合使用min.insync.replicas。它指定ISR中最少要有几个副本。比如你设置acksallmin.insync.replicas2副本数3那就意味着至少要有首领和一个跟随者都写成功才返回确认。这样即使首领挂了至少有一个跟随者有完整数据新的首领不会丢消息。但如果ISR里只剩一个副本首领自己那生产者的请求就会被拒绝因为达不到min.insync.replicas2的要求——宁可不可用也不能丢数据。消息批次的设计原理及参数调优生产者不会每条消息单独发一个网络请求太浪费了。它会在内存里把发往同一个分区的消息攒成一个batchbatch满了或者时间到了才一次性发出去。这是Kafka高性能的关键设计之一。涉及的核心参数有这么几个batch.size默认16KB。一个批次的最大字节数。设大一点批次更满网络请求次数更少吞吐更高。但同时也意味着要攒更多消息才发延迟会增加。高吞吐场景可以调到64KB甚至128KB。简单来说batch.size越大吞吐越高延迟越大自己根据业务做取舍。linger.ms默认0。意思是生产者发完一条消息后多等一会再发这段时间里如果有新的消息到了可以一起打包进batch。设0表示不等有消息就马上打包发出。设5~100ms能让消息多攒几条再发用一点点延迟换更高的吞吐。建议小流量场景设5~10ms高吞吐场景设20~50ms。batch.size和linger.ms是配合工作的。满足任何一个条件就发送要么batch大小达到了batch.size要么从第一条消息进入batch到现在过了linger.ms的时间。也就是说消息量大的时候batch.size先触发发送消息量小的时候linger.ms兜底不会让消息一直等。buffer.memory默认32MB。生产者用来攒batch的总内存大小。如果生产者的发送速度跟不上攒消息的速度缓冲池满了send()方法就会阻塞。高吞吐场景建议调到64MB~128MB。compression.type默认none。压缩算法可以设gzip、snappy、lz4、zstd。开了压缩之后消息在生产者端先压缩发给Broker的时候是压缩后的数据Broker直接存压缩后的数据消费者拉取的时候也是压缩后的数据消费者自己解压后再处理。从头到尾都省了带宽和磁盘空间。推荐用lz4或snappy这两个压缩解压速度最快虽然压缩率不是最高的但对Kafka这种追求吞吐的场景来说速度比压缩率重要。大消息发送的处理Kafka默认单条消息最大1MB。超过就会报RecordTooLargeException。虽然可以通过调参数把限制放大但一般不推荐这么干——大消息会带来一堆问题频繁GC大byte数组、网络拥塞一个大消息后面的小消息全排队等着、内存占用偏高、消费速度变慢等等。一般推荐这么处理第一选择是消息里只存引用。把文件或大内容存到OSS/HDFS/MinIO这些外部存储里Kafka消息里只带一个文件ID或者URL。消费者拿到后自己根据ID去外部存储取内容。这是最佳实践Kafka应该作为一个轻量级的消息管道不是文件存储系统。第二选择是消息分片。把一条大消息拆成多条小消息每条小消息带上原始消息ID 分片序号 总片数。消费者收到后根据序号组装还原。这比直接调大限制好但实现起来麻烦点。第三选择是调大所有相关参数。Producer的max.request.size、Broker的message.max.bytes和replica.fetch.max.bytes、Consumer的fetch.max.bytes这些参数都要调而且要保证Consumer端 ≥ Broker端 ≥ Producer端。简单粗暴但隐患多不到万不得已别这么干。retry机制生产者发送消息的时候如果遇到某些可重试的异常Kafka会自动重试。哪些异常会触发重试LeaderNotAvailableException分区Leader正在选举、NotEnoughReplicasExceptionISR副本不够、NetworkException网络抖动这些都是临时的重试有机会成功。哪些不会重试RecordTooLargeException消息太大了再怎么试也发不出去、SerializationException序列化问题不是网络问题、InvalidTopicExceptionTopic不存在这些重试没用。相关的参数retries是最大重试次数Kafka 2.1之后默认是Integer.MAX_VALUE无限重试但并不是真的无限重试因为有delivery.timeout.ms兜底——从第一次send到最终确认包括所有重试的时间总共不能超过delivery.timeout.ms默认120秒。retry.backoff.ms是两次重试之间的间隔时间默认100ms。重试机制有一个经典的坑重试可能导致消息乱序。举个例子两条消息msg1和msg2msg1先发送但发送失败进入重试msg2紧接着发送成功。此时Broker先收到了msg2msg1还在重试中。如果消费者这时候拉取拉到的顺序是msg2先于msg1——乱序了。对于订单状态变更这种依赖顺序的场景乱序会导致业务逻辑错误。解决方案是开启幂等性。enable.idempotence设成trueKafka 0.11可用。开启了之后Kafka会给每个生产者分配一个PIDProducer ID每条消息带一个递增的序列号Sequence Number。Broker收到消息后会检查同一个PID下这个消息的序列号是不是比已经确认的最大序号更大如果不是说明是重复发送或乱序到达直接丢弃。这样既保证了消息不重复又保证了顺序。powered by Kafka的幂等性原理本身很简单——靠PIDSeqNumber去重。它同时会自动把acks设成all把max.in.flight.requests.per.connection从默认的5降到5不需要降为1因为幂等性保证了顺序retriesInteger.MAX_VALUE。开了幂等性后性能几乎没影响因为max.in.flight.requests.per.connection不用降到1强烈建议默认开启。消费者及消费组消费者是Kafka里拉消息的那一端从Broker拉取消息来消费。为什么是拉Pull而不是推Push因为消费者自己最清楚自己的处理能力。如果是Push模式Broker拼命推消费者处理不过来消息就堆积在消费者内存里迟早被打爆。Pull模式下消费者按自己的节奏拉取能消化多少就拉多少。消费组是Kafka消费者最核心的概念它解决了怎么分摊消费压力的问题。同一个消费组内的多个消费者共同分摊主题的消费任务。具体来说一个分区只能被同一个组内的一个消费者消费。所以如果分区数是3、组内消费者也是3理想状态下每人一个分区。如果组内有4个消费者第4个是闲着的因为只有3个分区可以分。反过来如果有2个消费者、3个分区就有一个消费者要负责两个分区。这就得出一个结论分区数是消费者并行消费的上限。你想让消费能力翻倍得先让分区数够多消费者只是分配这些分区的单位。不同的消费组之间是完全独立的。同一个主题组A和组B各自维护自己的消费进度offset互不干扰。这也是Kafka能同时支撑消息队列模式和发布订阅模式的原因同一个组内的消费者之间是负载均衡消息队列模式不同组之间是广播发布订阅模式。消费组还有一个重要的机制叫Rebalance重平衡。当消费者加入组、离开组、长时间没心跳、订阅的主题分区数变化的时候Kafka会触发Rebalance重新分配分区和消费者的对应关系。关键是Rebalance期间整个消费组停止消费这就是所谓的惊群效应。触发Rebalance的常见原因中有两个参数很容易踩坑第一个是session.timeout.ms默认45秒消费者和Coordinator之间的心跳超时时间。心跳断了超过这个时间消费者就被认为挂了触发Rebalance。第二个是max.poll.interval.ms默认300秒两次poll()的最大间隔时间。有的业务处理比较慢比如一次poll拉500条、每条处理1秒500秒严重超300秒Kafka会认为消费者处理能力有问题把它踢出组触发Rebalance。解决方案要么是调小max.poll.records减少一次拉取的量要么调大max.poll.interval.ms。Rebalance很消耗性能应该尽量避免频繁发生。具体措施适当调大session.timeout.ms和max.poll.interval.ms根据业务处理时间来send让heartbeat能够正常发送。分区分配策略决定了分区怎么分给组内消费者。有几个策略Range是按主题粒度分配、RoundRobin是所有主题所有分区间轮询分配、Sticky尽量保持上次分配结果不变减少分区迁移、Cooperative Sticky更进一步它在Rebalance期间不需要全停——不需要重新分配的分区继续消费只处理有变化的部分增量式Rebalance。推荐用Cooperative Sticky。最后说一下Offset管理。Offset就是消费者消费到哪里的进度标记。消费者每次poll完一批消息后可以自动提交offsetenable.auto.committrue也可以手动提交设false。自动提交很简单不用写额外的代码Kafka后台定时auto.commit.interval.ms默认5秒帮你提交。但问题也在这poll来的消息你业务逻辑可能还没处理完offset已经提交了然后消费者挂了没处理完的那些消息就丢了因为下次启动时从已提交的offset往后读跳过了那些消息。手动提交是你自己控制什么时候提交一般是这批消息处理完再提交比较精确。手动提交又分两种commitSync同步提交阻塞等待直到提交成功可靠但慢和commitAsync异步提交不阻塞快但失败了不会自动重试。实际项目中通常处理完一批消息后异步提交生产者关闭前再同步提交一次确保最后的进度不丢while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecordString, String r : records) { processMessage(r); } // 处理完再异步提交 consumer.commitAsync((offsets, e) - { if (e ! null) { log.error(提交失败: {}, e.getMessage()); } }); } // 关闭前同步提交一次 try { consumer.commitSync(); } finally { consumer.close(); }另外有个参数auto.offset.reset要注意消费者第一次启动或者offset失效比如消息过期了的时候从哪开始读earliest是从最早的消息开始读从头消费latest是从最新的消息开始读只读新消息。消费者还有一些参数值得关注- max.poll.records一次poll最多拉多少条默认500。如果你的消息比较大或者处理比较耗时适当调小。- fetch.min.bytes一次拉取最少多少字节默认1。设大点可以减少空请求的次数。- fetch.max.wait.ms配合fetch.min.bytes使用数据不够fetch.min.bytes的时候最多等多久。- heartbeat.interval.ms心跳间隔默认3秒。跟session.timeout.ms配合一般session.timeout.ms设为heartbeat.interval.ms的3~4倍。