我開發(fā)了幾個(gè) C++ 應(yīng)用程序，它們生成和使用嵌入 Protobuf3 消息的 Kafka 消息（使用 cppkafka）。兩者都工作正常。生產(chǎn)者的相關(guān)代碼是：std::string kafkaString;cppkafka::MessageBuilder *builder;...solidList->SerializeToString(&kafkaString);builder->payload(kafkaString);Protobuf 對(duì)象被序列化為字符串并作為 Kafka 負(fù)載插入。到目前為止一切正?！，F(xiàn)在，我正在嘗試用 Java 開發(fā)一個(gè)消費(fèi)者。相關(guān)代碼應(yīng)該是：KafkaConsumer<Long, String> consumer=new KafkaConsumer<Long, String>(properties);....ConsumerRecords<Long, String> records = consumer.poll(100); for (ConsumerRecord<Long, String> record : records) { SolidList solidList = SolidList.parseFrom(record.value()); ...但在編譯時(shí)失?。?parseFrom 抱怨：類型 Solidlist.SolidList 中的方法 parseFrom(ByteBuffer) 不適用于參數(shù) (String)。所以，我嘗試使用 ByteBuffer：KafkaConsumer<Long, ByteBuffer> consumer=new KafkaConsumer<Long, ByteBuffer>(properties);....ConsumerRecords<Long, ByteBuffer> records = consumer.poll(100); for (ConsumerRecord<Long, ByteBuffer> record : records) { SolidList solidList = SolidList.parseFrom(record.value()); ...現(xiàn)在，錯(cuò)誤在執(zhí)行時(shí)間，仍然在 parseFrom(): Exception in thread "main" java.lang.ClassCastException: java.lang.String cannot be cast to java.nio.ByteBuffer。我知道它是一個(gè) java.lang.String ?。?！所以，我回到原來(lái)的狀態(tài)，并嘗試將其用作字節(jié)數(shù)組： SolidList solidList = SolidList.parseFrom(record.value().getBytes());現(xiàn)在，錯(cuò)誤出現(xiàn)在執(zhí)行時(shí)間：線程“main” com.google.protobuf.InvalidProtocolBufferException$InvalidWireTypeException 中的異常：協(xié)議消息標(biāo)記的線類型無(wú)效。.C++ 序列化的 protobuf 文檔說(shuō)明：bool SerializeToString(string output) const;：序列化消息并將字節(jié)存儲(chǔ)在給定的字符串中。請(qǐng)注意，字節(jié)是二進(jìn)制的，而不是文本；我們只使用字符串類作為方便的容器。*TL;DR：因此，我應(yīng)該如何解釋 Java 中的 protobuf C++“二進(jìn)制字節(jié)”？
2 回答
藍(lán)山帝景
TA貢獻(xiàn)1843條經(jīng)驗(yàn) 獲得超7個(gè)贊
嘗試實(shí)現(xiàn)一個(gè)反序列化器并將其作為值反序列化器傳遞給KafkaConsumer構(gòu)造函數(shù)。它可能看起來(lái)像這樣：
class SolidListDeserializer implements Deserializer<SolidList> {
public SolidList deserialize(final String topic, byte[] data) {
return SolidList.parseFrom(data);
}
...
}
...
KafkaConsumer<Long, SolidList> consumer = new KafkaConsumer<>(props, new LongDeserializer(), new SolidListDeserializer())
DIEA
TA貢獻(xiàn)1820條經(jīng)驗(yàn) 獲得超2個(gè)贊
您可以將 kafka 讀作ConsumerRecords<Long, String>. 接著SolidList.parseFrom(ByteBuffer.wrap(record.value().getBytes("UTF-8")));
添加回答
舉報(bào)
0/150
提交
取消