lywsvip/SpringBoot-Labs
1
0
Fork 0
mirror of https://gitee.com/yudaocode/SpringBoot-Labs.git synced 2026-09-12 10:17:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

增加 spring boot kafka 示例

Browse Source
This commit is contained in:
YunaiV
2019-12-07 08:53:54 +08:00
parent df49c52a1a
commit db82be9aba
6 changed files with 175 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lab-03/lab-03-kafka-demo-batch-consume/src/main/java/cn/iocoder/springboot/lab03/kafkademo/consumer/Demo02Consumer.java
View File

@@ -14,7 +14,7 @@ public class Demo02Consumer {
private Logger logger = LoggerFactory.getLogger(getClass());
@KafkaListener(topics = Demo02Message.TOPIC,
groupId = "demo02-B-consumer-group-" + Demo02Message.TOPIC)
groupId = "demo02-consumer-group-" + Demo02Message.TOPIC)
public void onMessage(List<Demo02Message> messages) {
logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size());
}

28
lab-03/lab-03-kafka-demo/src/main/java/cn/iocoder/springboot/lab03/kafkademo/config/KafkaConfiguration.java Normal file
View File

@@ -0,0 +1,28 @@
package cn.iocoder.springboot.lab03.kafkademo.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerRecordRecoverer;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.FixedBackOff;
@Configuration
public class KafkaConfiguration {
@Bean
@Primary
public ErrorHandler kafkaErrorHandler(KafkaTemplate<?, ?> template) {
// 创建 DeadLetterPublishingRecoverer 对象
ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
// 创建 FixedBackOff 对象
BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
// 创建 SeekToCurrentErrorHandler 对象
return new SeekToCurrentErrorHandler(recoverer, backOff);
}
}

42
lab-03/lab-03-kafka-demo/src/main/java/cn/iocoder/springboot/lab03/kafkademo/consumer/Demo04Consumer.java Normal file
View File

@@ -0,0 +1,42 @@
package cn.iocoder.springboot.lab03.kafkademo.consumer;
import cn.iocoder.springboot.lab03.kafkademo.message.Demo04Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.atomic.AtomicInteger;
@Component
public class Demo04Consumer {
private AtomicInteger count = new AtomicInteger(0);
private Logger logger = LoggerFactory.getLogger(getClass());
@KafkaListener(topics = Demo04Message.TOPIC,
groupId = "demo04-consumer-group-" + Demo04Message.TOPIC)
public void onMessage(Demo04Message message) {
logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
// 注意，此处抛出一个 RuntimeException 异常，模拟消费失败
throw new RuntimeException("我就是故意抛出一个异常");
}
// @Bean
// public ConsumerAwareListenerErrorHandler listenErrorHandler() {
// return new ConsumerAwareListenerErrorHandler() {
//
// @Override
// public Object handleError(Message<?> message,
// ListenerExecutionFailedException e,
// Consumer<?, ?> consumer) {
// System.out.println("message:" + message.getPayload());
// System.out.println("exception:" + e.getMessage());
// return null;
// }
//
// };
// }
}

31
lab-03/lab-03-kafka-demo/src/main/java/cn/iocoder/springboot/lab03/kafkademo/message/Demo04Message.java Normal file
View File

@@ -0,0 +1,31 @@
package cn.iocoder.springboot.lab03.kafkademo.message;
/**
* 示例 04 的 Message 消息
*/
public class Demo04Message {
public static final String TOPIC = "DEMO_04";
/**
* 编号
*/
private Integer id;
public Demo04Message setId(Integer id) {
this.id = id;
return this;
}
public Integer getId() {
return id;
}
@Override
public String toString() {
return "Demo01Message{" +
"id=" + id +
'}';
}
}

25
lab-03/lab-03-kafka-demo/src/main/java/cn/iocoder/springboot/lab03/kafkademo/producer/Demo04Producer.java Normal file
View File

@@ -0,0 +1,25 @@
package cn.iocoder.springboot.lab03.kafkademo.producer;
import cn.iocoder.springboot.lab03.kafkademo.message.Demo04Message;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
@Component
public class Demo04Producer {
@Resource
private KafkaTemplate<Object, Object> kafkaTemplate;
public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
// 创建 Demo04Message 消息
Demo04Message message = new Demo04Message();
message.setId(id);
// 同步发送消息
return kafkaTemplate.send(Demo04Message.TOPIC, message).get();
}
}

48
lab-03/lab-03-kafka-demo/src/test/java/cn/iocoder/springboot/lab03/kafkademo/producer/Demo04ProducerTest.java Normal file
View File

@@ -0,0 +1,48 @@
package cn.iocoder.springboot.lab03.kafkademo.producer;
import cn.iocoder.springboot.lab03.kafkademo.Application;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class Demo04ProducerTest {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private Demo04Producer producer;
@Test
public void testSyncSend() throws ExecutionException, InterruptedException {
int id = (int) (System.currentTimeMillis() / 1000);
SendResult result = producer.syncSend(id);
logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
// 阻塞等待，保证消费
new CountDownLatch(1).await();
}
@Test
public void testSyncSendX() throws ExecutionException, InterruptedException {
for (int i = 0; i < 100; i++) {
int id = (int) (System.currentTimeMillis() / 1000);
SendResult result = producer.syncSend(id);
logger.info("[testSyncSend][发送编号：[{}] 发送结果：[{}]]", id, result);
Thread.sleep(10 * 1000L);
}
// 阻塞等待，保证消费
new CountDownLatch(1).await();
}
}