Kafka是一個分布式的消息隊列系統,主要用于高吞吐量的數據傳輸。要傳輸二進制文件,可以將文件轉換為字節數組,并使用Producer API將字節數組發送到Kafka的Topic中。然后使用Consumer API從Kafka的Topic中接收字節數組,并將其轉換為二進制文件。
以下是一個使用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");
// 創建Producer
Producer<String, byte[]> producer = new KafkaProducer<>(props);
// 發送消息
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("消息發送成功,Topic: " + metadata.topic() +
", Partition: " + metadata.partition() +
", Offset: " + metadata.offset());
}
}
});
// 關閉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");
// 創建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) {
// 將字節數組轉換為二進制文件
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());
}
// 關閉Consumer
consumer.close();
}
}
請確保替換代碼中的path_to_file
和path_to_output_file
為合適的文件路徑。此外,還需要確保Kafka服務已經啟動并且主題binary-files-topic
已經創建。