kafka消费者java版本读取不到消息怎么办

Posted 2023-04-12

参考技术A 3.启动服务3.1启动zookeeper启动zk有两种方式，第一种是使用kafka自己带的一个zk。bin/zookeeper-server-start.shconfig/zookeeper.properties&另一种是使用其它的zookeeper，可以位于本机也可以位于其它地址。这种情况需要修改config下面的sercer.properties里面的zookeeper地址。例如zookeeper.connect=10.202.4.179:21813.2启动kafkabin/kafka-server-start.shconfig/server.properties4.创建topicbin/kafka-topics.sh--create--zookeeper10.202.4.179:2181--replication-factor1--partitions1--topictest创建一个名为test的topic，只有一个副本，一个分区。通过list命令查看刚刚创建的topicbin/kafka-topics.sh-list-zookeeper10.202.4.179:21815.启动producer并发送消息启动producerbin/kafka-console-producer.sh--broker-listlocalhost:9092--topictest启动之后就可以发送消息了比如testhelloboy按Ctrl+C退出发送消息6.启动consumerbin/kafka-console-consumer.sh--zookeeper10.202.4.179:2181--topictest--from-beginning启动consumer之后就可以在console中看到producer发送的消息了可以开启两个终端，一个发送消息，一个接受消息。如果这样都不行的话，查看zookeeper进程和kafka的topic，一步步排查原因吧。 参考技术B ：数据直接从通信网关过来？ 那最近每个map 得到的数据 是怎么区分的 是发数据端 按规则把数据配发到每个map？ 咱整过的一个例子是多个 map 同时从一张数据表取数 进行数据处理 在hdfs的输入目录 给每个map指定一个输入文件 map 读取这个输入文件

Kafka总结

参考技术A

consumer 采用 pull（拉）模式从 broker 中读取数据。
push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成 consumer 来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。
pull 模式不足之处是，如果 kafka 没有数据，消费者可能会陷入循环中，一直返回空数
据。针对这一点，Kafka 的消费者在消费数据时会传入一个时长参数 timeout，如果当前没有
数据可供消费，consumer 会等待一段时间之后再返回，这段时长即为 timeout。

分区中的所有副本统称为 AR（Assigned Replicas）。所有与 leader 副本保持一定程度同步的副本（包括 leader 副本在内）组成ISR（In-Sync Replicas），ISR 集合是 AR 集合中的一个子集。

可以通过分区策略体现消息顺序性。分区策略有轮询策略、随机策略、按消息键保序策略。

处理顺序 ：拦截器->序列化器->分区器
消息在通过 send() 方法发往 broker 的过程中，有可能需要经过拦截器（Interceptor）、序列化器（Serializer）和分区器（Partitioner）的一系列作用之后才能被真正地发往 broker。拦截器一般不是必需的，而序列化器是必需的。消息经过序列化之后就需要确定它发往的分区，如果消息 ProducerRecord 中指定了 partition 字段，那么就不需要分区器的作用，因为 partition 代表的就是所要发往的分区号。

整个生产者客户端由两个线程协调运行，这两个线程分别为主线程和 Sender 线程（发送线程）。

一般来说如果消费者过多，出现了消费者的个数大于分区个数的情况，就会有消费者分配不到任何分区。开发者可以继承AbstractPartitionAssignor实现自定义消费策略，从而实现同一消费组内的任意消费者都可以消费订阅主题的所有分区。

当前消费者需要提交的消费位移是offset+1
在旧消费者客户端中，消费位移是存储在 ZooKeeper 中的。而在新消费者客户端中，消费位移存储在 Kafka 内部的主题__consumer_offsets 中。

Kafka 中的消息是以主题为基本单位进行归类的，各个主题在逻辑上相互独立。每个主题又可以分为一个或多个分区。不考虑多副本的情况，一个分区对应一个日志（Log）。为了防止 Log 过大，Kafka 又引入了日志分段（LogSegment）的概念，将 Log 切分为多个 LogSegment，相当于一个巨型文件被平均分配为多个相对较小的文件。
Log 和 LogSegment 也不是纯粹物理意义上的概念，Log 在物理上只以文件夹的形式存储，而每个 LogSegment 对应于磁盘上的一个日志文件和两个索引文件，以及可能的其他文件。

每个日志分段文件对应了两个索引文件，主要用来提高查找消息的效率。

日志删除（Log Retention）：按照一定的保留策略直接删除不符合条件的日志分段。

日志压缩（Log Compaction）：针对每个消息的 key 进行整合，对于有相同 key 的不同 value 值，只保留最后一个版本。

在 Kafka 集群中会有一个或多个 broker，其中有一个 broker 会被选举为控制器（Kafka Controller），它负责管理整个集群中所有分区和副本的状态。当某个分区的 leader 副本出现故障时，由控制器负责为该分区选举新的 leader 副本。当检测到某个分区的 ISR 集合发生变化时，由控制器负责通知所有broker更新其元数据信息。当使用 kafka-topics.sh 脚本为某个 topic 增加分区数量时，同样还是由控制器负责分区的重新分配。

Kafka 中有多种延时操作，比如延时生产，还有延时拉取（DelayedFetch）、延时数据删除（DelayedDeleteRecords）等。
延时操作创建之后会被加入延时操作管理器（DelayedOperationPurgatory）来做专门的处理。延时操作有可能会超时，每个延时操作管理器都会配备一个定时器（SystemTimer）来做超时管理，定时器的底层就是采用时间轮（TimingWheel）实现的。

为了实现生产者的幂等性，Kafka 为此引入了 producer id（以下简称 PID）和序列号（sequence number）这两个概念。
Kafka的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer 在
初始化的时候会被分配一个 PID，发往同一 Partition 的消息会附带 Sequence Number。而Broker 端会对<PID, Partition, SeqNumber>做缓存，当具有相同主键的消息提交时，Broker 只会持久化一条。

Kafka中的事务可以使应用程序将消费消息、生产消息、提交消费位移当作原子操作来处理，同时成功或失败，即使该生产或消费会跨多个分区。

生产者必须提供唯一的transactionalId，启动后请求事务协调器获取一个PID，transactionalId与PID一一对应。

每次发送数据给<Topic, Partition>前，需要先向事务协调器发送AddPartitionsToTxnRequest，事务协调器会将该<Transaction, Topic, Partition>存于__transaction_state内，并将其状态置为BEGIN。

在处理完 AddOffsetsToTxnRequest 之后，生产者还会发送 TxnOffsetCommitRequest 请求给 GroupCoordinator，从而将本次事务中包含的消费位移信息 offsets 存储到主题 __consumer_offsets 中

一旦上述数据写入操作完成，应用程序必须调用KafkaProducer的commitTransaction方法或者abortTransaction方法以结束当前事务。

在发送延时消息的时候并不是先投递到要发送的真实主题（real_topic）中，而是先投递到一些 Kafka 内部的主题（delay_topic）中，这些内部主题对用户不可见，然后通过一个自定义的服务拉取这些内部主题中的消息，并将满足条件的消息再投递到要发送的真实的主题中，消费者所订阅的还是真实的主题。

Kafka 集群中有一个 broker 会被选举为 Controller，负责管理集群 broker 的上下线，所
有 topic 的分区副本分配和 leader 选举等工作。Controller 的管理工作都是依赖于 Zookeeper 的。