
36
|
第
3
章
需的。不过在遇到消息发送失败时,我们需要抛出异常、记录错误日志,或者把消息写入
“错误消息”文件以便日后分析。
为了在异步发送消息的同时能够对异常情况进行处理,生产者提供了回调支持。下面是使
用回调的一个例子。
private class DemoProducerCallback implements Callback {
➊
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
e.printStackTrace();
➋
}
}
}
ProducerRecord<String, String> record =
new ProducerRecord<>("CustomerCountry", "Biomedical Materials", "USA");
➌
producer.send(record, new DemoProducerCallback());
➍
➊
为了使用回调,需要一个实现了 org.apache.kafka.clients.producer.Callback 接口的
类,这个接口只有一个 onCompletion 方法。
➋
如果
Kafka
返回一个错误,onCompletion 方法会抛出一个非空(
non null
)异常。这里
我们只是简单地把它打印出来,但是在生产环境应该有更好的处理方式。
➌
记录与之前的一样。 ...