Kafka从入门到上天系列第二篇：直击生产者和消费者
一Kafka命令行操作1启动zk[rootlocalhost bin]# ./zkServer.sh start ZooKeeper JMX enabled by default Using config: /usr/local/src/zookeeper-3.4.14/bin/../conf/zoo.cfg Starting zookeeper ... STARTED [rootlocalhost bin]#2启动kafka[rootlocalhost bin]# ./kafka-server-start.sh -daemon ../config/server.properties [rootlocalhost logs]# tail -fn 100 server.log ssl.key.password null ssl.keymanager.algorithm SunX509 ssl.keystore.certificate.chain null ssl.keystore.key null ssl.keystore.location null ssl.keystore.password null ssl.keystore.type JKS ssl.principal.mapping.rules DEFAULT ssl.protocol TLSv1.2 ssl.provider null ssl.secure.random.implementation null ssl.trustmanager.algorithm PKIX ssl.truststore.certificates null ssl.truststore.location null ssl.truststore.password null ssl.truststore.type JKS telemetry.max.bytes 1048576 transaction.abort.timed.out.transaction.cleanup.interval.ms 10000 transaction.max.timeout.ms 900000 transaction.partition.verification.enable true transaction.remove.expired.transaction.cleanup.interval.ms 3600000 transaction.state.log.load.buffer.size 5242880 transaction.state.log.min.isr 1 transaction.state.log.num.partitions 50 transaction.state.log.replication.factor 1 transaction.state.log.segment.bytes 104857600 transactional.id.expiration.ms 604800000 unclean.leader.election.enable false unstable.api.versions.enable false unstable.metadata.versions.enable false zookeeper.clientCnxnSocket null zookeeper.connect 192.168.67.137:2181,192.168.67.138:2181,192.168.67.139:2181 zookeeper.connection.timeout.ms 18000 zookeeper.max.in.flight.requests 10 zookeeper.metadata.migration.enable false zookeeper.metadata.migration.min.batch.size 200 zookeeper.session.timeout.ms 18000 zookeeper.set.acl false zookeeper.ssl.cipher.suites null zookeeper.ssl.client.enable false zookeeper.ssl.crl.enable false zookeeper.ssl.enabled.protocols null zookeeper.ssl.endpoint.identification.algorithm HTTPS zookeeper.ssl.keystore.location null zookeeper.ssl.keystore.password null zookeeper.ssl.keystore.type null zookeeper.ssl.ocsp.enable false zookeeper.ssl.protocol TLSv1.2 zookeeper.ssl.truststore.location null zookeeper.ssl.truststore.password null zookeeper.ssl.truststore.type null (kafka.server.KafkaConfig) [2026-02-07 22:20:56,417] INFO [ThrottledChannelReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper) [2026-02-07 22:20:56,419] INFO [ThrottledChannelReaper-Produce]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper) [2026-02-07 22:20:56,419] INFO [ThrottledChannelReaper-Request]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper) [2026-02-07 22:20:56,426] INFO [ThrottledChannelReaper-ControllerMutation]: Starting (kafka.server.ClientQuotaManager$ThrottledChannelReaper) [2026-02-07 22:20:56,453] INFO Loading logs from log dirs ArrayBuffer(/usr/local/src/kafka_2.12-3.7.0/data/log) (kafka.log.LogManager) [2026-02-07 22:20:56,459] INFO No logs found to be loaded in /usr/local/src/kafka_2.12-3.7.0/data/log (kafka.log.LogManager) [2026-02-07 22:20:56,472] INFO Loaded 0 logs in 18ms (kafka.log.LogManager) [2026-02-07 22:20:56,478] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager) [2026-02-07 22:20:56,479] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager) [2026-02-07 22:20:56,800] INFO [kafka-log-cleaner-thread-0]: Starting (kafka.log.LogCleaner$CleanerThread) [2026-02-07 22:20:56,814] INFO [feature-zk-node-event-process-thread]: Starting (kafka.server.FinalizedFeatureChangeListener$ChangeNotificationProcessorThread) [2026-02-07 22:20:56,834] INFO [MetadataCache brokerId0] Updated cache from existing None to latest Features(version3.7-IV4, finalizedFeatures{}, finalizedFeaturesEpoch0). (kafka.server.metadata.ZkMetadataCache) [2026-02-07 22:20:56,860] INFO [zk-broker-0-to-controller-forwarding-channel-manager]: Starting (kafka.server.NodeToControllerRequestThread) [2026-02-07 22:20:57,177] INFO Updated connection-accept-rate max connection creation rate to 2147483647 (kafka.network.ConnectionQuotas) [2026-02-07 22:20:57,195] INFO [SocketServer listenerTypeZK_BROKER, nodeId0] Created>[rootlocalhost bin]# ./kafka-topics.sh --create \ --bootstrap-server localhost:9092 \ --topic test_topic \ --partitions 3 \ --replication-factor 1 \ --config retention.ms3600000 WARNING: Due to limitations in metric names, topics with a period (.) or underscore (_) could collide. To avoid issues it is best to use either, but not both. Created topic test_topic.4查看所有主题[rootlocalhost bin]# ./kafka-topics.sh --list --bootstrap-server localhost:9092 test_topic [rootlocalhost bin]#5查看主题详情[rootlocalhost bin]# ./kafka-topics.sh --describe \ --bootstrap-server localhost:9092 \ --topic test_topic Topic: test_topic TopicId: gfYhdnU9Qy2zdP_cvf3YVg PartitionCount: 3 ReplicationFactor: 1 Configs: retention.ms3600000 Topic: test_topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: test_topic Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Topic: test_topic Partition: 2 Leader: 0 Replicas: 0 Isr: 0输出会包含分区数、副本数、ISR同步副本、leader 分区等关键信息。6修改主题调整分区数 / 配置# 调整分区数注意分区数只能增加不能减少 ./kafka-topics.sh --alter \ --bootstrap-server localhost:9092 \ --topic test_topic \ --partitions 5 # 修改主题配置如修改保留时间为2小时 ./kafka-configs.sh --alter \ --bootstrap-server localhost:9092 \ --entity-type topics \ --entity-name test_topic \ --add-config retention.ms72000007删除主题bin/kafka-topics.sh --delete \ --bootstrap-server localhost:9092 \ --topic test_topic注意需确保server.properties中delete.topic.enabletrue默认开启否则仅标记为删除不会实际删除。命令行生产者和消费者当然也是可以通过命令行操作的但是这样真的有意义吗二生产者和消费者1生产者?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.dashu/groupId artifactIdkafka/artifactId version1.0-SNAPSHOT/version !--添加客户端-- dependencies dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version3.7.0/version /dependency /dependencies /projectpackage com.dashu; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; import java.util.concurrent.ExecutionException; public class KafkaProviderTest { public static void main(String[] args) { // 1. 配置生产者参数核心补充关键配置 Properties properties new Properties(); // Kafka集群地址 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.67.137:9092); // 显式配置序列化器推荐方式替代构造函数传参 properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // 消息确认级别1表示至少有一个副本确认接收保证消息不丢失 properties.put(ProducerConfig.ACKS_CONFIG, 1); // 发送失败重试次数 properties.put(ProducerConfig.RETRIES_CONFIG, 3); // 2. 创建生产者对象 KafkaProducerString, String kafkaProducer new KafkaProducer(properties); try { // 3. 循环发送消息同步发送确保每条都成功 for (int i 0; i 100; i) { System.out.println(正在发送第 i 条消息......); ProducerRecordString, String record new ProducerRecord( test_topic, // 目标Topic hello i, // 消息Key kafka i // 消息Value ); // 同步发送调用get()等待发送结果确保消息成功写入Kafka RecordMetadata metadata kafkaProducer.send(record).get(); // 打印发送结果可选验证消息发送到哪个分区/偏移量 System.out.printf(第 %d 条消息发送成功partition%d, offset%d%n, i, metadata.partition(), metadata.offset()); } } catch (InterruptedException | ExecutionException e) { // 捕获发送异常便于排查问题 System.err.println(消息发送失败 e.getMessage()); e.printStackTrace(); } finally { // 4. 优雅关闭生产者确保缓冲区剩余消息发送完成 kafkaProducer.flush(); // 刷新缓冲区 kafkaProducer.close(); System.out.println(生产者已关闭所有消息发送完成); } } }D:\soft\JDK\JDK21\bin\java.exe -javaagent:D:\soft\IntelliJ IDEA 2023.2.8\lib\idea_rt.jar11895:D:\soft\IntelliJ IDEA 2023.2.8\bin -Dfile.encodingUTF-8 -Dsun.stdout.encodingUTF-8 -Dsun.stderr.encodingUTF-8 -classpath D:\code\kafka\kafka\target\classes;D:\repository\org\apache\kafka\kafka-clients\3.7.0\kafka-clients-3.7.0.jar;D:\repository\com\github\luben\zstd-jni\1.5.5-6\zstd-jni-1.5.5-6.jar;D:\repository\org\lz4\lz4-java\1.8.0\lz4-java-1.8.0.jar;D:\repository\org\xerial\snappy\snappy-java\1.1.10.5\snappy-java-1.1.10.5.jar;D:\repository\org\slf4j\slf4j-api\1.7.36\slf4j-api-1.7.36.jar com.dashu.KafkaProviderTest SLF4J: Failed to load class org.slf4j.impl.StaticLoggerBinder. SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 正在发送第 0 条消息...... 第 0 条消息发送成功partition0, offset47 正在发送第 1 条消息...... 第 1 条消息发送成功partition2, offset32 正在发送第 2 条消息...... 第 2 条消息发送成功partition2, offset33 正在发送第 3 条消息...... 第 3 条消息发送成功partition0, offset48 正在发送第 4 条消息...... 第 4 条消息发送成功partition2, offset34 正在发送第 5 条消息...... 第 5 条消息发送成功partition0, offset49 正在发送第 6 条消息...... 第 6 条消息发送成功partition1, offset21 正在发送第 7 条消息...... 第 7 条消息发送成功partition0, offset50 正在发送第 8 条消息...... 第 8 条消息发送成功partition0, offset51 正在发送第 9 条消息...... 第 9 条消息发送成功partition1, offset22 正在发送第 10 条消息...... 第 10 条消息发送成功partition0, offset52 正在发送第 11 条消息...... 第 11 条消息发送成功partition2, offset35 正在发送第 12 条消息...... 第 12 条消息发送成功partition2, offset36 正在发送第 13 条消息...... 第 13 条消息发送成功partition1, offset23 正在发送第 14 条消息...... 第 14 条消息发送成功partition1, offset24 正在发送第 15 条消息...... 第 15 条消息发送成功partition0, offset53 正在发送第 16 条消息...... 第 16 条消息发送成功partition0, offset54 正在发送第 17 条消息...... 第 17 条消息发送成功partition0, offset55 正在发送第 18 条消息...... 第 18 条消息发送成功partition1, offset25 正在发送第 19 条消息...... 第 19 条消息发送成功partition1, offset26 正在发送第 20 条消息...... 第 20 条消息发送成功partition0, offset56 正在发送第 21 条消息...... 第 21 条消息发送成功partition0, offset57 正在发送第 22 条消息...... 第 22 条消息发送成功partition2, offset37 正在发送第 23 条消息...... 第 23 条消息发送成功partition0, offset58 正在发送第 24 条消息...... 第 24 条消息发送成功partition2, offset38 正在发送第 25 条消息...... 第 25 条消息发送成功partition2, offset39 正在发送第 26 条消息...... 第 26 条消息发送成功partition2, offset40 正在发送第 27 条消息...... 第 27 条消息发送成功partition0, offset59 正在发送第 28 条消息...... 第 28 条消息发送成功partition1, offset27 正在发送第 29 条消息...... 第 29 条消息发送成功partition0, offset60 正在发送第 30 条消息...... 第 30 条消息发送成功partition2, offset41 正在发送第 31 条消息...... 第 31 条消息发送成功partition1, offset28 正在发送第 32 条消息...... 第 32 条消息发送成功partition1, offset29 正在发送第 33 条消息...... 第 33 条消息发送成功partition2, offset42 正在发送第 34 条消息...... 第 34 条消息发送成功partition0, offset61 正在发送第 35 条消息...... 第 35 条消息发送成功partition2, offset43 正在发送第 36 条消息...... 第 36 条消息发送成功partition2, offset44 正在发送第 37 条消息...... 第 37 条消息发送成功partition0, offset62 正在发送第 38 条消息...... 第 38 条消息发送成功partition2, offset45 正在发送第 39 条消息...... 第 39 条消息发送成功partition2, offset46 正在发送第 40 条消息...... 第 40 条消息发送成功partition1, offset30 正在发送第 41 条消息...... 第 41 条消息发送成功partition1, offset31 正在发送第 42 条消息...... 第 42 条消息发送成功partition2, offset47 正在发送第 43 条消息...... 第 43 条消息发送成功partition2, offset48 正在发送第 44 条消息...... 第 44 条消息发送成功partition0, offset63 正在发送第 45 条消息...... 第 45 条消息发送成功partition1, offset32 正在发送第 46 条消息...... 第 46 条消息发送成功partition0, offset64 正在发送第 47 条消息...... 第 47 条消息发送成功partition0, offset65 正在发送第 48 条消息...... 第 48 条消息发送成功partition0, offset66 正在发送第 49 条消息...... 第 49 条消息发送成功partition0, offset67 正在发送第 50 条消息...... 第 50 条消息发送成功partition0, offset68 正在发送第 51 条消息...... 第 51 条消息发送成功partition2, offset49 正在发送第 52 条消息...... 第 52 条消息发送成功partition1, offset33 正在发送第 53 条消息...... 第 53 条消息发送成功partition2, offset50 正在发送第 54 条消息...... 第 54 条消息发送成功partition2, offset51 正在发送第 55 条消息...... 第 55 条消息发送成功partition0, offset69 正在发送第 56 条消息...... 第 56 条消息发送成功partition0, offset70 正在发送第 57 条消息...... 第 57 条消息发送成功partition0, offset71 正在发送第 58 条消息...... 第 58 条消息发送成功partition1, offset34 正在发送第 59 条消息...... 第 59 条消息发送成功partition2, offset52 正在发送第 60 条消息...... 第 60 条消息发送成功partition2, offset53 正在发送第 61 条消息...... 第 61 条消息发送成功partition0, offset72 正在发送第 62 条消息...... 第 62 条消息发送成功partition2, offset54 正在发送第 63 条消息...... 第 63 条消息发送成功partition0, offset73 正在发送第 64 条消息...... 第 64 条消息发送成功partition2, offset55 正在发送第 65 条消息...... 第 65 条消息发送成功partition2, offset56 正在发送第 66 条消息...... 第 66 条消息发送成功partition0, offset74 正在发送第 67 条消息...... 第 67 条消息发送成功partition0, offset75 正在发送第 68 条消息...... 第 68 条消息发送成功partition0, offset76 正在发送第 69 条消息...... 第 69 条消息发送成功partition1, offset35 正在发送第 70 条消息...... 第 70 条消息发送成功partition0, offset77 正在发送第 71 条消息...... 第 71 条消息发送成功partition0, offset78 正在发送第 72 条消息...... 第 72 条消息发送成功partition0, offset79 正在发送第 73 条消息...... 第 73 条消息发送成功partition0, offset80 正在发送第 74 条消息...... 第 74 条消息发送成功partition0, offset81 正在发送第 75 条消息...... 第 75 条消息发送成功partition1, offset36 正在发送第 76 条消息...... 第 76 条消息发送成功partition0, offset82 正在发送第 77 条消息...... 第 77 条消息发送成功partition0, offset83 正在发送第 78 条消息...... 第 78 条消息发送成功partition2, offset57 正在发送第 79 条消息...... 第 79 条消息发送成功partition0, offset84 正在发送第 80 条消息...... 第 80 条消息发送成功partition0, offset85 正在发送第 81 条消息...... 第 81 条消息发送成功partition0, offset86 正在发送第 82 条消息...... 第 82 条消息发送成功partition1, offset37 正在发送第 83 条消息...... 第 83 条消息发送成功partition1, offset38 正在发送第 84 条消息...... 第 84 条消息发送成功partition2, offset58 正在发送第 85 条消息...... 第 85 条消息发送成功partition1, offset39 正在发送第 86 条消息...... 第 86 条消息发送成功partition0, offset87 正在发送第 87 条消息...... 第 87 条消息发送成功partition0, offset88 正在发送第 88 条消息...... 第 88 条消息发送成功partition1, offset40 正在发送第 89 条消息...... 第 89 条消息发送成功partition2, offset59 正在发送第 90 条消息...... 第 90 条消息发送成功partition0, offset89 正在发送第 91 条消息...... 第 91 条消息发送成功partition0, offset90 正在发送第 92 条消息...... 第 92 条消息发送成功partition0, offset91 正在发送第 93 条消息...... 第 93 条消息发送成功partition2, offset60 正在发送第 94 条消息...... 第 94 条消息发送成功partition1, offset41 正在发送第 95 条消息...... 第 95 条消息发送成功partition2, offset61 正在发送第 96 条消息...... 第 96 条消息发送成功partition2, offset62 正在发送第 97 条消息...... 第 97 条消息发送成功partition0, offset92 正在发送第 98 条消息...... 第 98 条消息发送成功partition2, offset63 正在发送第 99 条消息...... 第 99 条消息发送成功partition0, offset93 生产者已关闭所有消息发送完成 Process finished with exit code 02消费者?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.dashu/groupId artifactIdkafka/artifactId version1.0-SNAPSHOT/version !--添加客户端-- dependencies dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version3.7.0/version /dependency /dependencies /projectpackage com.dashu; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class KafkaConsumerTest { public static void main(String[] args) { Properties properties new Properties(); // 集群地址和生产者一致 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.67.137:9092); // 消费组ID properties.put(ConsumerConfig.GROUP_ID_CONFIG, test_group); // 显式配置反序列化器 properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 关键从头消费能收到历史消息 properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); // 创建消费者对象 KafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties); // 订阅目标Topic kafkaConsumer.subscribe(Collections.singletonList(test_topic)); System.out.println(消费者已启动开始监听 test_topic...); try { // 循环消费 while (true) { // 新版poll方法用Duration指定超时时间1秒 ConsumerRecordsString, String records kafkaConsumer.poll(Duration.ofSeconds(1)); if (records.isEmpty()) { // 无消息时提示避免控制台空白 System.out.println(暂未收到新消息继续监听...); continue; } // 遍历消费到的消息 for (ConsumerRecordString, String record : records) { System.out.printf(收到消息key%s, value%s, partition%d, offset%d%n, record.key(), record.value(), record.partition(), record.offset()); } } } catch (Exception e) { System.err.println(消费异常 e.getMessage()); e.printStackTrace(); } finally { // 关闭消费者实际运行时可通过信号量退出循环后执行 kafkaConsumer.close(); System.out.println(消费者已关闭); } } }收到消息keyhello0, valuekafka0, partition0, offset47 收到消息keyhello1, valuekafka1, partition2, offset32 收到消息keyhello2, valuekafka2, partition2, offset33 收到消息keyhello3, valuekafka3, partition0, offset48 收到消息keyhello4, valuekafka4, partition2, offset34 收到消息keyhello5, valuekafka5, partition0, offset49 收到消息keyhello6, valuekafka6, partition1, offset21 收到消息keyhello7, valuekafka7, partition0, offset50 收到消息keyhello8, valuekafka8, partition0, offset51 收到消息keyhello9, valuekafka9, partition1, offset22 收到消息keyhello10, valuekafka10, partition0, offset52 收到消息keyhello11, valuekafka11, partition2, offset35 收到消息keyhello12, valuekafka12, partition2, offset36 收到消息keyhello13, valuekafka13, partition1, offset23 收到消息keyhello14, valuekafka14, partition1, offset24 收到消息keyhello15, valuekafka15, partition0, offset53 收到消息keyhello16, valuekafka16, partition0, offset54 收到消息keyhello17, valuekafka17, partition0, offset55 收到消息keyhello18, valuekafka18, partition1, offset25 收到消息keyhello19, valuekafka19, partition1, offset26 收到消息keyhello20, valuekafka20, partition0, offset56 收到消息keyhello21, valuekafka21, partition0, offset57 收到消息keyhello22, valuekafka22, partition2, offset37 收到消息keyhello23, valuekafka23, partition0, offset58 收到消息keyhello24, valuekafka24, partition2, offset38 收到消息keyhello25, valuekafka25, partition2, offset39 收到消息keyhello26, valuekafka26, partition2, offset40 收到消息keyhello27, valuekafka27, partition0, offset59 收到消息keyhello28, valuekafka28, partition1, offset27 收到消息keyhello29, valuekafka29, partition0, offset60 收到消息keyhello30, valuekafka30, partition2, offset41 收到消息keyhello31, valuekafka31, partition1, offset28 收到消息keyhello32, valuekafka32, partition1, offset29 收到消息keyhello33, valuekafka33, partition2, offset42 收到消息keyhello34, valuekafka34, partition0, offset61 收到消息keyhello35, valuekafka35, partition2, offset43 收到消息keyhello36, valuekafka36, partition2, offset44 收到消息keyhello37, valuekafka37, partition0, offset62 收到消息keyhello38, valuekafka38, partition2, offset45 收到消息keyhello39, valuekafka39, partition2, offset46 收到消息keyhello40, valuekafka40, partition1, offset30 收到消息keyhello41, valuekafka41, partition1, offset31 收到消息keyhello42, valuekafka42, partition2, offset47 收到消息keyhello43, valuekafka43, partition2, offset48 收到消息keyhello44, valuekafka44, partition0, offset63 收到消息keyhello45, valuekafka45, partition1, offset32 收到消息keyhello46, valuekafka46, partition0, offset64 收到消息keyhello47, valuekafka47, partition0, offset65 收到消息keyhello48, valuekafka48, partition0, offset66 收到消息keyhello49, valuekafka49, partition0, offset67 收到消息keyhello50, valuekafka50, partition0, offset68 收到消息keyhello51, valuekafka51, partition2, offset49 收到消息keyhello53, valuekafka53, partition2, offset50 收到消息keyhello52, valuekafka52, partition1, offset33 收到消息keyhello54, valuekafka54, partition2, offset51 收到消息keyhello55, valuekafka55, partition0, offset69 收到消息keyhello56, valuekafka56, partition0, offset70 收到消息keyhello57, valuekafka57, partition0, offset71 收到消息keyhello58, valuekafka58, partition1, offset34 收到消息keyhello59, valuekafka59, partition2, offset52 收到消息keyhello60, valuekafka60, partition2, offset53 收到消息keyhello61, valuekafka61, partition0, offset72 收到消息keyhello62, valuekafka62, partition2, offset54 收到消息keyhello63, valuekafka63, partition0, offset73 收到消息keyhello64, valuekafka64, partition2, offset55 收到消息keyhello65, valuekafka65, partition2, offset56 收到消息keyhello66, valuekafka66, partition0, offset74 收到消息keyhello67, valuekafka67, partition0, offset75 收到消息keyhello68, valuekafka68, partition0, offset76 收到消息keyhello69, valuekafka69, partition1, offset35 收到消息keyhello70, valuekafka70, partition0, offset77 收到消息keyhello71, valuekafka71, partition0, offset78 收到消息keyhello72, valuekafka72, partition0, offset79 收到消息keyhello73, valuekafka73, partition0, offset80 收到消息keyhello74, valuekafka74, partition0, offset81 收到消息keyhello75, valuekafka75, partition1, offset36 收到消息keyhello76, valuekafka76, partition0, offset82 收到消息keyhello77, valuekafka77, partition0, offset83 收到消息keyhello78, valuekafka78, partition2, offset57 收到消息keyhello79, valuekafka79, partition0, offset84 收到消息keyhello80, valuekafka80, partition0, offset85 收到消息keyhello81, valuekafka81, partition0, offset86 收到消息keyhello82, valuekafka82, partition1, offset37 收到消息keyhello83, valuekafka83, partition1, offset38 收到消息keyhello84, valuekafka84, partition2, offset58 收到消息keyhello85, valuekafka85, partition1, offset39 收到消息keyhello86, valuekafka86, partition0, offset87 收到消息keyhello87, valuekafka87, partition0, offset88 收到消息keyhello88, valuekafka88, partition1, offset40 收到消息keyhello89, valuekafka89, partition2, offset59 收到消息keyhello90, valuekafka90, partition0, offset89 收到消息keyhello91, valuekafka91, partition0, offset90 收到消息keyhello92, valuekafka92, partition0, offset91 收到消息keyhello93, valuekafka93, partition2, offset60 收到消息keyhello94, valuekafka94, partition1, offset41 收到消息keyhello95, valuekafka95, partition2, offset61 收到消息keyhello96, valuekafka96, partition2, offset62 收到消息keyhello97, valuekafka97, partition0, offset92 收到消息keyhello98, valuekafka98, partition2, offset63 收到消息keyhello99, valuekafka99, partition0, offset93 暂未收到新消息继续监听... 暂未收到新消息继续监听... 暂未收到新消息继续监听... 暂未收到新消息继续监听... 暂未收到新消息继续监听...
