Kafka无法接收二进制数据

h43kikqp  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(566)

我正试图发送一个二进制格式的音频剪辑到Kafka主题。
但Kafka没有收到这条信息。
以下是我的制作人:

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.log4j.BasicConfigurator;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

public class AudioProducer {

public static void main(String[] args) {
    BasicConfigurator.configure();
    System.out.println("program started");
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "broker-host:9092");
    properties.put("acks", "all");
    properties.put("retries", 0);
    properties.put("batch.size", 26214400);
    properties.put("linger.ms", 1);
    properties.put("buffer.memory", 2*26214400);
    properties.put("max.request.size", 26214400);
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    KafkaProducer<String,byte[]> producer = new KafkaProducer<String, byte[]>(properties);
    try {
        byte[] temp =Files.readAllBytes(Paths.get(args[0]));
        System.out.println("input path:"+args[0]);
        producer.send(new ProducerRecord<String,byte[]>("audio-queue", "test-key",temp ));
    } catch (IOException e) {
        e.printStackTrace();
    }
    producer.close();
    System.out.println("program completed");
}

}

以下是kafka调试模式的输出:

program started
0 [main] INFO org.apache.kafka.clients.producer.ProducerConfig  - ProducerConfig values: 
    compression.type = none
    metric.reporters = []
    metadata.max.age.ms = 300000
    metadata.fetch.timeout.ms = 60000
    acks = all
    batch.size = 26214400
    reconnect.backoff.ms = 10
    bootstrap.servers = [broker-host:9092]
    receive.buffer.bytes = 32768
    retry.backoff.ms = 100
    buffer.memory = 52428800
    timeout.ms = 30000
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    retries = 0
    max.request.size = 26214400
    block.on.buffer.full = true
    value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
    metrics.sample.window.ms = 30000
    send.buffer.bytes = 131072
    max.in.flight.requests.per.connection = 5
    metrics.num.samples = 2
    linger.ms = 1
    client.id = 

86 [main] DEBUG org.apache.kafka.clients.producer.internals.Metadata  - Updated cluster metadata version 1 to Cluster(nodes = [Node(broker-host, 9092)], partitions = [])
105 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender  - Starting Kafka producer I/O thread.
106 [main] DEBUG org.apache.kafka.clients.producer.KafkaProducer  - Kafka producer started
input path:AUD_0030.wav
190 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Trying to send metadata request to node -1
190 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Init connection to node -1 for sending metadata request in the next iteration
190 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Initiating connection to node -1 at broker-host:9092.
251 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Trying to send metadata request to node -1
261 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Completed connection to node -1
351 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Trying to send metadata request to node -1
361 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Sending metadata request ClientRequest(expectResponse=true, payload=null, request=RequestSend(header={api_key=3,api_version=0,correlation_id=0,client_id=producer-1}, body={topics=[audio-queue]})) to node -1
977 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Metadata  - Updated cluster metadata version 2 to Cluster(nodes = [Node(1, broker-host, 9092)], partitions = [Partition(topic = audio-queue, partition = 0, leader = 1, replicas = [1,], isr = [1,]])
1021 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender  - Beginning shutdown of Kafka producer I/O thread, sending remaining records.
1021 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Initiating connection to node 1 at 01hw508208.india.tcs.com:9092.
1037 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.NetworkClient  - Completed connection to node 1
11511 [kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.producer.internals.Sender  - Shutdown of Kafka producer I/O thread has completed.
11512 [main] DEBUG org.apache.kafka.clients.producer.KafkaProducer  - The Kafka producer has closed.
program completed

但是相同的主题和程序可以很好地处理字符串消息。
我还检查了broker节点上的kafka日志。我只能找到字符串消息,但找不到二进制消息。

tuwxkamq

tuwxkamq1#

由于它解决了问题,我将用我的评论作为回答。
kafka不是一个文件服务器,在处理千字节范围内的消息时性能最好。默认情况下,最大消息大小为1 mb,可以通过将brokers max.message.bytes属性设置为更高的值来覆盖。
这样做的结果是,应该通过以下方式增加使用者的最大获取量(在新的使用者api中) fetch.max.bytes 大消息有性能缺陷。当发送更大的文件时,应该考虑将文件存储在存储系统(例如s3)上,并且只将uri传递给kafka中的那些文件。

相关问题