@Test public void test() { kafkaProducer.sendTransactionMsg("test", "上联：天王盖地虎"); kafkaProducer.sendTransactionMsg("test", "下联：宝塔镇河妖"); }
@Transactional(rollbackFor = RuntimeException.class) public void sendTransactionMsg(String topic, String data) { log.info("向kafka发送数据:[{}]", data); template.executeInTransaction(t -> { t.send(topic, "prepare"); if ("error".equals(data)) { throw new RuntimeException("failed"); } t.send(topic, "finish"); return true; }); }
KafkaProducer { @Transactional(rollbackFor = RuntimeException.class) public void sendTransactionMsg(String topic, String data) { log.info("向kafka发送数据:[{}]", data); template.executeInTransaction(t -> { t.send(topic, "prepare"); if ("error".equals(data)) { throw new RuntimeException("failed"); } t.send(topic, "finish"); return true; }); } }
KafkaProducer { @Transactional(rollbackFor = RuntimeException.class) public void sendTransactionMsg(String topic, String data) { log.info("向kafka发送数据:[{}]", data); template.executeInTransaction(t -> { t.send(topic, "prepare"); if ("error".equals(data)) { throw new RuntimeException("failed"); } t.send(topic, "finish"); return true; }); }  }
KafkaProducer { @Transactional(rollbackFor = RuntimeException.class) public void sendTransactionMsg(String topic, String data) { log.info("向kafka发送数据:[{}]", data); template.executeInTransaction(t -> { t.send(topic, "prepare"); if ("error".equals(data)) { throw new RuntimeException("failed"); } t.send(topic, "finish"); return true; }); }  @Transactional(rollbackFor = RuntimeException.class) void sendTransactionMsg(String topic, String data); }
KafkaProducer { @Transactional(rollbackFor = RuntimeException.class) public void sendTransactionMsg(String topic, String data) { log.info("向kafka发送数据:[{}]", data); template.executeInTransaction(t -> { t.send(topic, "prepare"); if ("error".equals(data)) { throw new RuntimeException("failed"); } t.send(topic, "finish"); return true; }); }  @Transactional(rollbackFor = RuntimeException.class) void sendTransactionMsg(String topic, String data);  }