是否可以使用 Kafka 传输文件？答案

【问题标题】：Is it possible to transfer files using Kafka?是否可以使用 Kafka 传输文件？
【发布时间】：2016-12-30 22:23:53
【问题描述】：

我每天都会生成数千个文件，我想使用 Kafka 进行流式传输。当我尝试读取文件时，每一行都被视为一条单独的消息。

我想知道如何将每个文件的内容作为 Kafka 主题中的一条消息，以及如何将来自 Kafka 主题的每条消息写入单独的文件中。

【问题讨论】：

你看过 Kafka Connect 吗？ docs.confluent.io/3.0.0/connect/index.html
是的，我知道。我怎样才能在这里使用它？场景是当我读取文件时，每一行都被视为一条单独的消息，但我希望每个文件都是一条长消息。（文件可能有 30-40 行）
您使用的是 Java 客户端、控制台生产者还是其他？
是的，我的制作人将主要使用 Java，但我也愿意接受其他选择。
嗨@Nahush 你能把你用来实现这个场景的代码发给我吗！我找不到任何参考资料如何为这种场景编写制作人

标签： apache-kafka kafka-consumer-api kafka-producer-api

【解决方案1】：

您可以编写自己的序列化器/反序列化器来处理文件。例如：

制片人道具：

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);  
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, YOUR_FILE_SERIALIZER_URI);

消费道具：

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, YOUR_FILE_DESERIALIZER_URI);

序列化器

public class FileMapSerializer implements Serializer<Map<?,?>> {

@Override
public void close() {

}

@Override
public void configure(Map configs, boolean isKey) {
}

@Override
public byte[] serialize(String topic, Map data) {
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    ObjectOutput out = null;
    byte[] bytes = null;
    try {
        out = new ObjectOutputStream(bos);
        out.writeObject(data);
        bytes = bos.toByteArray();
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        try {
            if (out != null) {
                out.close();
            }
        } catch (IOException ex) {
            // ignore close exception
        }
        try {
            bos.close();
        } catch (IOException ex) {
            // ignore close exception
        }
    }
    return bytes;
}
}

反序列化器

public class MapDeserializer implements Deserializer<Map> {

@Override
public void close() {

}

@Override
public void configure(Map config, boolean isKey) {

}

@Override
public Map deserialize(String topic, byte[] message) {
    ByteArrayInputStream bis = new ByteArrayInputStream(message);
    ObjectInput in = null;
    try {
        in = new ObjectInputStream(bis);
        Object o = in.readObject();
        if (o instanceof Map) {
            return (Map) o;
        } else
            return new HashMap<String, String>();
    } catch (ClassNotFoundException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        try {
            bis.close();
        } catch (IOException ex) {
        }
        try {
            if (in != null) {
                in.close();
            }
        } catch (IOException ex) {
            // ignore close exception
        }
    }
    return new HashMap<String, String>();
}
}

按以下格式编写消息

final Object kafkaMessage = new ProducerRecord<String, Map>((String) <TOPIC>,Integer.toString(messageId++), messageMap);

messageMap 将包含文件名作为键和文件内容作为值。值可以是可序列化的对象。因此，每条消息都将包含一个带有 File_Name 与 FileContent 映射的映射。可以是单个值或多个值。

【讨论】：