干货1、kafka客户端图形工具如CMAK、Kafka-map等上看到的消息不管实际队列中有没有压缩展示到界面上的都是解压后的如果存储的是压缩后的数据工具会自动解压后展示到界面上2、命令kafka-topics.sh --describe *** 看到消息信息不一定能看到实际的压缩信息验证不了是否被压缩3、命令kafka-console-consumer.sh *** 的 print-storage-info 参数在kafka3.4版本开始就废弃了所以之后的版本无法使用该参数验证其是否被压缩4、验证压缩最权威的方法就是直接读取 Broker 上的日志文件展示每条消息的压缩状态在Apache Kafka中数据压缩是一种常见的优化手段特别是在传输大量数据时。压缩可以显著减少网络带宽的使用提高数据传输的效率。Kafka支持多种压缩格式如GZIP、Snappy、LZ4等。验证Kafka队列中的数据是否被压缩可以通过以下几种方法来实现1. 使用Kafka命令行工具Kafka提供了kafka-console-consumer命令行工具可以用来查看队列中的数据。你可以在消费数据时指定压缩格式以便查看是否启用了压缩。例如如果你知道消息使用了GZIP压缩你可以在消费时指定压缩格式kafka-console-consumer --bootstrap-server broker-list --topic topic-name --from-beginning --property print.keytrue --property print.valuetrue --property print.timestamptrue --property key.deserializerorg.apache.kafka.common.serialization.StringDeserializer --property value.deserializerorg.apache.kafka.common.serialization.ByteArrayDeserializer --property compression.typegzip2. 使用Kafka API如果你正在使用Kafka的Java客户端或其他编程语言客户端如Python、C#等你可以在配置消费者时指定压缩格式并检查接收到的消息是否已经解压。Java 示例Properties props new Properties(); props.put(bootstrap.servers, broker-list); props.put(group.id, test); props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(value.deserializer, org.apache.kafka.common.serialization.ByteArrayDeserializer); props.put(compression.type, gzip); // 如果已知压缩类型 KafkaConsumerString, byte[] consumer new KafkaConsumer(props); consumer.subscribe(Arrays.asList(topic-name)); while (true) { ConsumerRecordsString, byte[] records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, byte[] record : records) { // 在这里你可以检查record的值是否是你期望的解压后的数据格式 System.out.printf(offset %d, key %s, value %s%n, record.offset(), record.key(), new String(record.value())); } }3. 检查Kafka主题配置你也可以检查Kafka主题的配置看看是否启用了压缩。可以通过Kafka命令行工具kafka-configs来查看主题的配置kafka-configs --bootstrap-server broker-list --entity-type topics --entity-name topic-name --describe | grep compression.type这将显示该主题的压缩类型设置。4. 查看Broker日志或监控工具在某些情况下你也可以查看Kafka broker的日志或使用监控工具如Kafka Manager, Confluent Control Center等来查看主题的配置和性能指标间接确认是否使用了压缩。通过上述方法你可以有效地验证Kafka队列中的数据是否被压缩并了解如何处理或消费这些压缩后的数据。附件一查看Broker日志验证举例./kafka-dump-log.sh \ --files /private/tmp/kraft-combined-logs/skywalking-logs-0/00000000000000000000.log \ --print-data-log下面是作者在MacOS上通过查看Broker日志验证过程举例附件二Kafka可视化管理工具-)Kafka Map -- 国产、开源通过配置kafka地址实现管理作者使用kafka-map: 一个美观简洁且强大的kafka web管理工具。-)kafka CMAK-- 通过配置zookeeper地址实现管理作者使用GitHub - yahoo/CMAK: CMAK is a tool for managing Apache Kafka clusters-)kafka-monitor-- 开源https://github.com/linkedin/kafka-monitor/wiki-)kafka-eagle-- 商业软件收费EFAK等等附件三几种压缩算法对比附件四数据压缩--图表总结附件五Kafka几项核心的底层技术Kafka 的高性能、高可靠性和高吞吐量源于其一系列精妙的底层设计技术。以下是几项核心的底层技术1. 分区Partition与并行处理Kafka 将每个主题Topic划分为多个分区每个分区是一个有序的、不可变的消息日志。这种设计是 Kafka 实现高吞吐量和水平扩展的基础。并行读写生产者可以将消息并行写入不同的分区消费者也可以并行地从多个分区拉取消息极大地提升了系统的并发处理能力。负载均衡分区可以分布在集群中的不同 Broker 上实现了数据的分布式存储和负载均衡。顺序性保证Kafka 保证了分区内消息的严格顺序虽然跨分区的消息顺序无法保证但对于大多数场景分区内有序已足够。2. 顺序写入与零拷贝Zero-CopyKafka 充分利用了磁盘和操作系统的特性来优化性能。顺序写入消息被追加Append到分区日志文件的末尾这种顺序写入方式避免了磁盘频繁的寻道操作使得 Kafka 在普通机械硬盘上也能实现极高的写入吞吐量。零拷贝技术当消费者读取消息时Kafka 利用操作系统的sendfile系统调用直接将数据从磁盘文件缓冲区传输到网络缓冲区避免了数据在内核态与用户态之间的多次拷贝。这显著降低了 CPU 开销提升了网络传输效率。3. 副本机制Replication与 ISR为了保证数据的高可用性和可靠性Kafka 引入了副本机制。Leader/Follower每个分区有多个副本其中一个被选举为 Leader负责处理所有生产者和消费者的读写请求其他副本为 Follower仅从 Leader 异步拉取数据进行同步。ISRIn-Sync ReplicasKafka 维护一个“同步副本”列表ISR只有那些与 Leader 副本同步延迟在可接受范围内的 Follower 才会被包含在 ISR 中。只有当消息被写入 ISR 列表中的所有副本后生产者才会收到确认ack-1这确保了即使 Leader 宕机数据也不会丢失。高可用性当 Leader 宕机时Kafka 会从 ISR 列表中选举出一个新的 Leader保证服务不中断。4. 批量处理与消息压缩Kafka 通过批量操作来 amortize分摊网络和磁盘 I/O 开销。批量发送生产者会将多条消息缓存起来达到一定大小batch.size或等待一定时间linger.ms后再一次性发送给 Broker减少了网络请求次数。消息压缩生产者可以在发送前对消息批次进行压缩如 gzip、snappy、lz4减少了网络传输的数据量和磁盘存储空间。Broker 在存储和传输时也以压缩格式处理消费者在消费时再解压。5. 分段日志Segment与索引为了管理庞大的日志文件并实现高效的消息查找Kafka 采用了分段和索引机制。分段Segment每个分区的日志文件被切割成多个大小固定的段Segment文件。当一个段文件达到预设大小如 1GB时Kafka 会创建一个新的段文件。这使得日志文件的管理和清理如删除过期数据更加高效。索引文件每个段文件对应两个索引文件——偏移量索引.index和时间戳索引.timeindex。偏移量索引采用稀疏存储方式通过二分查找可以快速定位到消息在日志文件中的物理位置实现了 O(log n) 的消息查找时间复杂度。6. 消费者位移Offset管理Kafka 将消费进度即位移的管理权交给了消费者而非 Broker。消费者自主管理消费者在消费消息后会定期将自己消费到的位移Offset提交并保存可保存在 Kafka 内部主题__consumer_offsets或外部存储如 Redis 中。容错与重放如果消费者宕机重启后可以从最后一次提交的位移处继续消费保证了“至少一次”At-Least-Once的语义。这种设计也允许消费者灵活地重放历史消息。