1
카프카를 사용하여 메시지를 보내고 메시지를 보내고 있습니다.Kafka ByteArray
제작자는 <String, ByteArray> 제작자와 함께 작업해도 좋습니다.
소비 할 때, 아래 코드를 사용하고 있지만 각 레코드는 단지 8 바이트 (코드 아래의 샘플 출력)로 표시됩니다.
소비자가 단순히 전체 메시지를 바이트 배열로 가져갈 수있는 방법이 있습니까?
코드 : System.out에의
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(props);
consumer.subscribe(Arrays.asList(topic));
int i = 0;
while (true) {
ConsumerRecords<String, byte[]> records = consumer.poll(100);
for (ConsumerRecord<String, byte[]> record : records)
System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
}
출력 :
offset = 1773133, key = 105906453, value = [[email protected]
offset = 1773134, key = 105906453, value = [[email protected]
offset = 1773135, key = 105906453, value = [[email protected]
offset = 1773136, key = 105906453, value = [[email protected]
offset = 1773137, key = 105906177, value = [[email protected]
offset = 1773138, key = 105906177, value = [[email protected]
offset = 1773139, key = 105906177, value = [[email protected]
offset = 1773140, key = 105906177, value = [[email protected]
offset = 1773141, key = 105906177, value = [[email protected]
내가 생각으로 메시지를 다시 만들려면 다음 기록의 부하를 조립하는 데 찾고 아니에요 뭔가 빠졌고 수동 조립이 잘못 될 수 있습니다.
저주! 나는 지금까지 내 머리를 카프카에 올려 놓았지만, 나는 이것을 놓쳤다. 그러나 나는 아직도 나의 제작자가 생산하고있는 각각의 메시지 하나 하나에 대해 여러 개의 청크/오프셋/이름을 얻고있다. –
들어오는 각 메시지에 대해 하나의 바이트 배열을 가져 오는 일종의 리스너/루프를 사용할 수 없습니까? –
자신이하는 일은 간단합니다.지도의 각 요소에 대해 메서드를 호출하면됩니다. KafkaConsumer는 여전히 상당히 낮은 수준의 API이므로 동일한 배치에서 다시 호출되므로 실제 프로토콜에 데이터가 표시됩니다. 메시지 당 하나의 콜백으로 추상화하는 것을 원한다면 Samza (http://samza.apache.org/) 나 KafkaStreams (http://www.confluent.io/blog/)와 같은 것을 조사해야합니다. 소개 - 카프카 - 스트림 - 스트림 처리 - 만든 - 간단한 /) –