我在本地机器上有一个消息生产者,远程主机(aws)上有一个代理。
从生产者发送消息后,我等待并调用远程主机上的控制台消费者,发现有过多的日志记录,但没有生产者的值。
生产者在调用发送方法后刷新数据。一切都已正确配置。
我该如何检查代理是否收到了来自生产者的消息,并查看生产者是否收到了回复?
我在本地机器上有一个消息生产者,远程主机(aws)上有一个代理。
从生产者发送消息后,我等待并调用远程主机上的控制台消费者,发现有过多的日志记录,但没有生产者的值。
生产者在调用发送方法后刷新数据。一切都已正确配置。
我该如何检查代理是否收到了来自生产者的消息,并查看生产者是否收到了回复?
Send
方法异步地将消息发送到主题,并返回一个Future
对象,其中包含RecordMetadata
。
java.util.concurrent.Future<RecordMetadata> send(ProducerRecord<K,V> record)
异步地将记录发送到主题
在调用flush
方法之后,通过调用isDone
方法检查Future是否已完成(例如,Future.isDone() == true
)。
调用此方法将使所有缓冲的记录立即可用于发送(即使linger.ms大于0),并阻塞等待与这些记录相关联的请求完成。flush()的后置条件是任何先前发送的记录都已完成(例如,Future.isDone() == true)。当请求根据您指定的acks配置成功确认或导致错误时,请求被视为已完成。
RecordMetadata
包含offset
和partition
public int partition()
记录被发送到的分区
public long offset()
记录的偏移量,如果{hasOffset()}返回false,则为-1。
或者您也可以使用Callback函数来确保消息是否已发送到主题
完全非阻塞使用可以利用回调参数提供回调,在请求完成时将调用该回调。
这里有一个清晰的示例在文档中
ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
producer.send(myRecord,
new Callback() {
public void onCompletion(RecordMetadata metadata, Exception e) {
if(e != null) {
e.printStackTrace();
} else {
System.out.println("The offset of the record we just sent is: " + metadata.offset());
}
}
});
ProducerRecord<String, String> record =
new ProducerRecord<>("SampleTopic", "SampleKey", "SampleValue");
try {
producer.send(record).get();
} catch (Exception e) {
e.printStackTrace();
}