lywsvip/SpringBoot-Labs
1
0
Fork 0
mirror of https://gitee.com/yudaocode/SpringBoot-Labs.git synced 2026-09-03 05:53:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

增加 spring boot kafka 示例

Browse Source
This commit is contained in:
YunaiV
2019-12-07 18:32:31 +08:00
parent e559e7c45c
commit d6abdcd079
2 changed files with 26 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
lab-03/lab-03-kafka-demo-concurrency/src/main/java/cn/iocoder/springboot/lab03/kafkademo/producer/Demo06Producer.java
View File

@@ -22,4 +22,13 @@ public class Demo06Producer {
return kafkaTemplate.send(Demo06Message.TOPIC, message).get();
}
public SendResult syncSendOrderly(Integer id) throws ExecutionException, InterruptedException {
// 创建 Demo01Message 消息
Demo06Message message = new Demo06Message();
message.setId(id);
// 同步发送消息
// 因为我们使用 String 的方式序列化 key ，所以需要将 id 转换成 String
return kafkaTemplate.send(Demo06Message.TOPIC, String.valueOf(id), message).get();
}
}

20
lab-03/lab-03-kafka-demo-concurrency/src/test/java/cn/iocoder/springboot/lab03/kafkademo/producer/Demo06ProducerTest.java
View File

@@ -24,9 +24,23 @@ public class Demo06ProducerTest {
@Test
public void testSyncSend() throws ExecutionException, InterruptedException {
int id = (int) (System.currentTimeMillis() / 1000);
SendResult result = producer.syncSend(id);
logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
for (int i = 0; i < 10; i++) {
int id = (int) (System.currentTimeMillis() / 1000);
SendResult result = producer.syncSend(id);
// logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
}
// 阻塞等待，保证消费
new CountDownLatch(1).await();
}
@Test
public void testSyncSendOrderly() throws ExecutionException, InterruptedException {
for (int i = 0; i < 10; i++) {
int id = 1;
SendResult result = producer.syncSendOrderly(id);
logger.info("[testSyncSend][发送编号：[{}] 发送队列：[{}]]", id, result.getRecordMetadata().partition());
}
// 阻塞等待，保证消费
new CountDownLatch(1).await();