@Test public void testASyncSend() throws InterruptedException { int id = (int) (System.currentTimeMillis() / 1000); producer.asyncSend(id).addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() { @Override public void onFailure(Throwable e) { logger.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e); } @Override public void onSuccess(SendResult<Object, Object> result) { logger.info("[testASyncSend][发送编号：[{}] 发送成功，结果为：[{}]]", id, result); } }); new CountDownLatch(1).await(); }
public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) { Demo01Message message = new Demo01Message(); message.setId(id); return kafkaTemplate.send(Demo01Message.TOPIC, message); }
Demo01Producer { public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) { Demo01Message message = new Demo01Message(); message.setId(id); return kafkaTemplate.send(Demo01Message.TOPIC, message); } }
Demo01Producer { public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) { Demo01Message message = new Demo01Message(); message.setId(id); return kafkaTemplate.send(Demo01Message.TOPIC, message); }  }
Demo01Producer { public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) { Demo01Message message = new Demo01Message(); message.setId(id); return kafkaTemplate.send(Demo01Message.TOPIC, message); }  SendResult syncSend(Integer id); ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id); }
Demo01Producer { public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id) { Demo01Message message = new Demo01Message(); message.setId(id); return kafkaTemplate.send(Demo01Message.TOPIC, message); }  SendResult syncSend(Integer id); ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id);  }