java - 是否可以使用 java 将 kafka 消费者接收到的输出写入文件

Question

我有一个用 java 编写的 Kafka Producer 代码，用于编写 kafka 消息。以及接收这些消息的消费者代码。

是否可以将消费者收到的这些消息写入java中的任何文本文件。

Answer 1

如果您正在编写自己的使用者，您应该在同一个应用程序中包含写入文件的逻辑。使用预打包的控制台使用者，您可以将其通过管道传输到文件中。例如：kafka-console-consumer > file.txt

另一个（无代码）选项是尝试StreamSets Data Collector，这是一个开源的 Apache 许可工具，它也有一个拖放 UI。它包括用于 Kafka 的内置连接器和各种数据格式。

*完全披露我是这个项目的提交者。

Answer 2

多谢你们，

我能够实现它。一旦在消费者端接收到数据，那么它只是您必须编写的常见 java 代码。

下面是 ode 中将消息打印到控制台的行。

System.out.println(String.valueOf(messageAndOffset.offset()) + ": " + new String(bytes, "UTF-8"));

您可以将所有消息存储到字符串中，并一次将所有消息打印到文件中。

System.out.println(String.valueOf(messageAndOffset.offset()) + ": " + new String(bytes, "UTF-8"));
completMessage += new String(bytes, "UTF-8")+"\n";

new String(bytes, "UTF-8")+"\n";包含实际消息。

最后将所有消息打印到文件中。

writeDataToFile(completMessage);

writeDataToFile包含将字符串写入文件的简单 java 代码。

谢谢你。

Answer 3

有可能的。下面是这个的工作代码。

package com.venk.prac;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;

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.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import kafka.utils.ShutdownableThread;

public class FileConsumer extends ShutdownableThread {

    private final KafkaConsumer<Integer, String> kafkaConsumer;
    private String topic;
    private String filePath;
    private BufferedWriter buffWriter;

    public FileConsumer(String topic, String filePath) {

        super("FileConsumer", false);
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                KafkaProperties.KAFKA_BROKER_SERVERS_PORTS_STRING);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "FileConsumer");
        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        properties.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        kafkaConsumer = new KafkaConsumer<Integer, String>(properties);
        this.topic = topic;
        this.filePath = filePath;

        try {
            this.buffWriter = new BufferedWriter(new FileWriter(filePath));
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public void doWork() {
        // TODO Auto-generated method stub
        kafkaConsumer.subscribe(Collections.singletonList(this.topic));
        ConsumerRecords<Integer, String> consumerRecords = kafkaConsumer.poll(1000);
        try {
            for (ConsumerRecord<Integer, String> record : consumerRecords) 
                buffWriter.write(record.value() + System.lineSeparator());
            buffWriter.flush();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public String name() {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public boolean isInterruptible() {
        return false;
    }

}