溫馨提示×

kafka怎么傳輸二進制文件

小億
141
2023-10-27 15:39:55

Kafka是一個分布式的消息隊列系統(tǒng)，主要用于高吞吐量的數(shù)據(jù)傳輸。要傳輸二進制文件，可以將文件轉(zhuǎn)換為字節(jié)數(shù)組，并使用Producer API將字節(jié)數(shù)組發(fā)送到Kafka的Topic中。然后使用Consumer API從Kafka的Topic中接收字節(jié)數(shù)組，并將其轉(zhuǎn)換為二進制文件。

以下是一個使用Java的示例代碼：

Producer端：

import org.apache.kafka.clients.producer.*;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;

public class FileProducer {
    private final static String TOPIC = "binary-files-topic";
    private final static String BOOTSTRAP_SERVERS = "localhost:9092";

    public static void main(String[] args) throws IOException {
        // 讀取二進制文件
        File file = new File("path_to_file");
        byte[] data = new byte[(int)file.length()];
        FileInputStream fis = new FileInputStream(file);
        fis.read(data);
        fis.close();

        // Kafka Producer配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");

        // 創(chuàng)建Producer
        Producer<String, byte[]> producer = new KafkaProducer<>(props);

        // 發(fā)送消息
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(TOPIC, "key", data);
        producer.send(record, new Callback() {
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    System.out.println("消息發(fā)送成功，Topic: " + metadata.topic() +
                            ", Partition: " + metadata.partition() +
                            ", Offset: " + metadata.offset());
                }
            }
        });

        // 關(guān)閉Producer
        producer.close();
    }
}

Consumer端：

import org.apache.kafka.clients.consumer.*;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;

public class FileConsumer {
    private final static String TOPIC = "binary-files-topic";
    private final static String BOOTSTRAP_SERVERS = "localhost:9092";
    private final static String OUTPUT_FILE = "path_to_output_file";

    public static void main(String[] args) throws IOException {
        // Kafka Consumer配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "binary-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");

        // 創(chuàng)建Consumer
        Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);

        // 訂閱Topic
        consumer.subscribe(Collections.singletonList(TOPIC));

        // 接收消息
        ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, byte[]> record : records) {
            // 將字節(jié)數(shù)組轉(zhuǎn)換為二進制文件
            File outputFile = new File(OUTPUT_FILE);
            FileOutputStream fos = new FileOutputStream(outputFile);
            fos.write(record.value());
            fos.close();
            System.out.println("接收到消息，Topic: " + record.topic() +
                    ", Partition: " + record.partition() +
                    ", Offset: " + record.offset());
        }

        // 關(guān)閉Consumer
        consumer.close();
    }
}

請確保替換代碼中的path_to_file和path_to_output_file為合適的文件路徑。此外，還需要確保Kafka服務(wù)已經(jīng)啟動并且主題binary-files-topic已經(jīng)創(chuàng)建。

0 踩