kafka消费者关闭再次打开如何显示生产者的数据

在 Kafka 中，消费者是可以随时关闭再次打开的。如果你关闭了消费者，再次启动它时，它将从上一次消费的偏移量处继续消费消息。如果你想要查看生产者发送的消息，你需要在消费者启动之后，重新发送这些消息到 Kafka 的一个新的主题中，然后使用一个新的消费者来消费这个新的主题。

具体而言，你可以使用 Kafka Connect 来创建一个 Sink Connector，将原始主题中的消息发送到一个新的主题中。然后，你可以使用一个新的消费者来消费这个新的主题中的消息。这样，你就可以在消费者关闭再次打开时，查看之前生产者发送的所有消息了。

另外，如果你在生产者发送消息时，将消息的键设置为一个固定的值，那么你可以使用 Kafka Streams 库来过滤出所有具有该键的消息。具体而言，你可以使用 KTable 来将原始主题中的消息转换为一个键值对，其中键是消息的键，值是消息本身。然后，你可以使用 KStream 来过滤出所有具有特定键的消息，并将它们发送到一个新的主题中。最后，你可以使用一个新的消费者来消费这个新的主题中的消息。这种方法需要编写一些代码，但是可以更加灵活地处理消息的过滤和转换。

相关问题

kafka关闭生产者消费者

Kafka生产者可以在发送完消息后调用close()方法关闭，示例代码如下：

producer.close();

而Kafka消费者则需要调用close()方法关闭消费者和其它资源，示例代码如下：

consumer.close();

hadoop平台中kafka关闭生产者消费者

在Hadoop平台中关闭Kafka生产者和消费者的步骤如下：

1. 停止生产者：使用命令行工具进入Kafka服务器的安装目录，使用以下命令停止生产者：
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test --property "parse.key=true" --property "key.separator=:"
CTRL+C

2. 停止消费者：使用命令行工具进入Kafka服务器的安装目录，使用以下命令停止消费者：
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
CTRL+C

3. 停止Kafka服务器：在命令行工具中输入以下命令停止Kafka服务器：
bin/kafka-server-stop.sh

以上步骤是在命令行中操作的，如果使用集成开发环境，也可以在IDE中停止生产者和消费者。