分析:该问题发生主要原因是由于生产者的生产能力大于了kafka server的接收落盘能力,导致生产者本地队列写满,默认情况下,如果生产者本地缓冲队列写满会抛出“Local: Queue full”异常。
如何解决这个异常呢?
上面将了,根因是生产者的活干的有点快。所以需要让本地队列里面的数据消化一会儿poll()。
直接上代码吧:
retry:
RdKafka::ErrorCode err = producer->produce(
topic,
RdKafka::Topic::PARTITION_UA,
RdKafka::Producer::RK_MSG_COPY ,
const_cast(line.c_str()), line.size(),
NULL, 0,
0,
NULL,
NULL);
if (err != RdKafka::ERR_NO_ERROR) {
std::cerr << "% Failed to produce to topic " << topic << ": "
<< RdKafka::err2str(err) << std::endl;
if (err == RdKafka::ERR__QUEUE_FULL) {
producer->poll(1000 );
goto retry;
}
} else {
std::cerr << "% Enqueued message (" << line.size() << " bytes) "
<< "for topic " << topic << std::endl;
}
注意:这里捕获异常ERR__QUEUE_FULL)后使用了poll(1000),就是让本地队列里面的数据向kafka发送一会儿(1000ms)。



