@Test public void testSyncSendInTransaction() throws ExecutionException, InterruptedException { int id = (int) (System.currentTimeMillis() / 1000); producer.syncSendInTransaction(id, new Runnable() { @Override public void run() { logger.info("[run][我要开始睡觉了]"); try { Thread.sleep(10 * 1000L); } catch (InterruptedException e) { throw new RuntimeException(e); } logger.info("[run][我睡醒了]"); } }); new CountDownLatch(1).await(); }
public String syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException { return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() { @Override public String doInOperations(KafkaOperations<Object, Object> kafkaOperations) { Demo07Message message = new Demo07Message(); message.setId(id); try { SendResult<Object, Object> sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get(); logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult); } catch (Exception e) { throw new RuntimeException(e); } runner.run(); return "success"; } }); }
Demo07Producer { public String syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException { return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() { @Override public String doInOperations(KafkaOperations<Object, Object> kafkaOperations) { Demo07Message message = new Demo07Message(); message.setId(id); try { SendResult<Object, Object> sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get(); logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult); } catch (Exception e) { throw new RuntimeException(e); } runner.run(); return "success"; } }); } }
Demo07Producer { public String syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException { return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() { @Override public String doInOperations(KafkaOperations<Object, Object> kafkaOperations) { Demo07Message message = new Demo07Message(); message.setId(id); try { SendResult<Object, Object> sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get(); logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult); } catch (Exception e) { throw new RuntimeException(e); } runner.run(); return "success"; } }); }  }
Demo07Producer { public String syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException { return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() { @Override public String doInOperations(KafkaOperations<Object, Object> kafkaOperations) { Demo07Message message = new Demo07Message(); message.setId(id); try { SendResult<Object, Object> sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get(); logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult); } catch (Exception e) { throw new RuntimeException(e); } runner.run(); return "success"; } }); }  SendResult syncSend(Integer id); String syncSendInTransaction(Integer id, Runnable runner); }
Demo07Producer { public String syncSendInTransaction(Integer id, Runnable runner) throws ExecutionException, InterruptedException { return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() { @Override public String doInOperations(KafkaOperations<Object, Object> kafkaOperations) { Demo07Message message = new Demo07Message(); message.setId(id); try { SendResult<Object, Object> sendResult = kafkaOperations.send(Demo07Message.TOPIC, message).get(); logger.info("[doInOperations][发送编号：[{}] 发送结果：[{}]]", id, sendResult); } catch (Exception e) { throw new RuntimeException(e); } runner.run(); return "success"; } }); }  SendResult syncSend(Integer id); String syncSendInTransaction(Integer id, Runnable runner);  }