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已经创建。
