为什么卡夫卡消费者不听第一条消息？

在我的测试程序中，我开始收听。然后在循环中发送消息。如果我发送一条消息，它不列出该消息。如果我发送2条消息，它将监听一条消息。如果我发送3条消息，它会收听2条消息。为什么？为什么卡夫卡消费者不听第一条消息？

生产者

KeyedMessage<String, byte[]> message = new KeyedMessage<String, byte[]>(topic, serializedBytes); 

if (log.isDebugEnabled()) { 
    log.debug("producing messages to topic : " + topic + "file : " + payload.get("name")); 
} 

for (int i = 0; i < 3; i++) { 
    producer.send(message); 
    System.out.println("producing .."); 
}

消费者

public void run() { 

    try { 
     ConsumerIterator<byte[], byte[]> itr = m_stream.iterator(); 
     log.info("Kafka listener is ready to listen.."); 
     System.out.println("listens...."); 

     while (itr.hasNext()) { 
      byte[] data = itr.next().message(); 
      System.out.println("Message received : " + data); 
     } 
    } 
}

消费者属性

enable.auto.commit=true 

auto.commit.interval.ms=101 

session.timeout.ms=7000 

key.deserializer=org.apache.kafka.common.serialization.StringDeserializer 

zookeeper.connect=zk1.xx\:2181 

heartbeat.interval.ms=1000 

auto.offset.reset=smallest 

serializer.class=kafka.serializer.DefaultEncoder 

bootstrap.servers=kk1.xx\:9092 

group.id=test 

consumer.timeout.ms=-1 

fetch.min.bytes=1 

receive.buffer.bytes=262144

来源

2016-05-02 Ratha

你在你的kafka主题中配置了多少个分区？ – jbarrueta

@jbarrueta它是默认主题..0分区 – Ratha

我由s固定此在我的制作人中关注财产。

request.required.acks=1

来源

2016-05-03 01:26:27 Ratha

为什么卡夫卡消费者不听第一条消息？

回答

相关问题