diff --git a/lab-40/lab-40-kafka/pom.xml b/lab-40/lab-40-kafka/pom.xml new file mode 100644 index 00000000..6050f8d9 --- /dev/null +++ b/lab-40/lab-40-kafka/pom.xml @@ -0,0 +1,68 @@ + + + + org.springframework.boot + spring-boot-starter-parent + 2.1.11.RELEASE + + + 4.0.0 + + lab-40-kafka + + + + + + org.springframework.kafka + spring-kafka + 2.2.11.RELEASE + + + + + org.springframework.boot + spring-boot-starter-web + + + + + + io.zipkin.brave + brave + + + io.zipkin.reporter2 + zipkin-sender-okhttp3 + + + + + + io.zipkin.brave + brave-instrumentation-spring-webmvc + + + + io.zipkin.brave + brave-instrumentation-kafka-clients + + + + + + + + + io.zipkin.brave + brave-bom + 5.9.1 + pom + import + + + + + diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/KafkaApplication.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/KafkaApplication.java new file mode 100644 index 00000000..0213566a --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/KafkaApplication.java @@ -0,0 +1,13 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class KafkaApplication { + + public static void main(String[] args) { + SpringApplication.run(KafkaApplication.class, args); + } + +} diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/config/SpringMvcConfiguration.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/config/SpringMvcConfiguration.java new file mode 100644 index 00000000..09fd2000 --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/config/SpringMvcConfiguration.java @@ -0,0 +1,25 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo.config; + +import brave.spring.webmvc.SpanCustomizingAsyncHandlerInterceptor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.web.servlet.config.annotation.InterceptorRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; + +@Configuration +@Import(SpanCustomizingAsyncHandlerInterceptor.class) // 创建拦截器 SpanCustomizingAsyncHandlerInterceptor Bean +public class SpringMvcConfiguration implements WebMvcConfigurer { + + @Autowired + public SpanCustomizingAsyncHandlerInterceptor webMvcTracingCustomizer; + + /** + * Decorates server spans with application-defined web tags + */ + @Override + public void addInterceptors(InterceptorRegistry registry) { // 记录 SpringMVC 相关信息到 Span 中 + registry.addInterceptor(webMvcTracingCustomizer); + } + +} diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/config/ZipkinConfiguration.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/config/ZipkinConfiguration.java new file mode 100644 index 00000000..93c74652 --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/config/ZipkinConfiguration.java @@ -0,0 +1,138 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo.config; + +import brave.CurrentSpanCustomizer; +import brave.SpanCustomizer; +import brave.Tracing; +import brave.http.HttpTracing; +import brave.kafka.clients.KafkaTracing; +import brave.servlet.TracingFilter; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.ProducerFactory; +import zipkin2.Span; +import zipkin2.reporter.AsyncReporter; +import zipkin2.reporter.Sender; +import zipkin2.reporter.okhttp3.OkHttpSender; + +import javax.servlet.Filter; +import java.util.Properties; + +@Configuration +public class ZipkinConfiguration { + + // ==================== 通用配置 ==================== + + /** + * Configuration for how to send spans to Zipkin + */ + @Bean + public Sender sender() { // Sender 采用 HTTP 通信方式 + return OkHttpSender.create("http://127.0.0.1:9411/api/v2/spans"); + } + + /** + * Configuration for how to buffer spans into messages for Zipkin + */ + @Bean + public AsyncReporter spanReporter() { // 异步 Reporter + return AsyncReporter.create(sender()); + } + + /** + * Controls aspects of tracing such as the service name that shows up in the UI + */ + @Bean + public Tracing tracing(@Value("${spring.application.name}") String serviceName) { + return Tracing.newBuilder() + .localServiceName(serviceName) // 应用名 + .spanReporter(this.spanReporter()).build(); + } + + /** + * Allows someone to add tags to a span if a trace is in progress + */ + @Bean + public SpanCustomizer spanCustomizer(Tracing tracing) { + return CurrentSpanCustomizer.create(tracing); + } + + // ==================== HTTP 相关 ==================== + + /** + * Decides how to name and tag spans. By default they are named the same as the http method + */ + @Bean + public HttpTracing httpTracing(Tracing tracing) { + return HttpTracing.create(tracing); + } + + /** + * Creates server spans for http requests + */ + @Bean + public Filter tracingFilter(HttpTracing httpTracing) { // 拦截请求，记录 HTTP 请求的链路信息 + return TracingFilter.create(httpTracing); + } + + // ==================== SpringMVC 相关 ==================== + // @see SpringMvcConfiguration 类上的，@Import(SpanCustomizingAsyncHandlerInterceptor.class) 。因为 SpanCustomizingAsyncHandlerInterceptor 未提供 public 构造方法 + + // ==================== Kafka 相关 ==================== + + @Bean + public KafkaTracing kafkaTracing(Tracing tracing) { + return KafkaTracing.newBuilder(tracing) + .remoteServiceName("demo-mq-kafka") + .build(); + } + + @Bean + public ProducerFactory kafkaProducerFactory(KafkaProperties properties, KafkaTracing kafkaTracing) { + DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory(properties.buildProducerProperties()) { + + @Override + public Producer createProducer() { + // 创建默认的 Producer + Producer producer = super.createProducer(); + // 创建可链路追踪的 Producer + return kafkaTracing.producer(producer); + } + + }; + + String transactionIdPrefix = properties.getProducer().getTransactionIdPrefix(); + if (transactionIdPrefix != null) { + factory.setTransactionIdPrefix(transactionIdPrefix); + } + + return factory; + } + + @Bean + public ConsumerFactory kafkaConsumerFactory(KafkaProperties properties, KafkaTracing kafkaTracing) { + return new DefaultKafkaConsumerFactory(properties.buildConsumerProperties()) { + + @Override + public Consumer createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix) { + return this.createConsumer(groupId, clientIdPrefix, clientIdSuffix, null); + } + + @Override + public Consumer createConsumer(String groupId, String clientIdPrefix, final String clientIdSuffixArg, Properties properties) { + // 创建默认的 Consumer + Consumer consumer = super.createConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties); + // 创建可链路追踪的 Consumer + return kafkaTracing.consumer(consumer); + } + + }; + } + +} diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/consumer/DemoConsumer.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/consumer/DemoConsumer.java new file mode 100644 index 00000000..ccb1b38f --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/consumer/DemoConsumer.java @@ -0,0 +1,20 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo.consumer; + +import cn.iocoder.springboot.lab40.skywalkingdemo.message.DemoMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +@Component +public class DemoConsumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + @KafkaListener(topics = DemoMessage.TOPIC, + groupId = "demo-consumer-group-" + DemoMessage.TOPIC) + public void onMessage(DemoMessage message) { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + } + +} diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/controller/DemoController.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/controller/DemoController.java new file mode 100644 index 00000000..3d89fdb3 --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/controller/DemoController.java @@ -0,0 +1,28 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo.controller; + +import cn.iocoder.springboot.lab40.skywalkingdemo.producer.DemoProducer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.concurrent.ExecutionException; + +@RestController +@RequestMapping("/demo") +public class DemoController { + + @Autowired + private DemoProducer producer; + + @GetMapping("/kafka") + public String echo() throws ExecutionException, InterruptedException { + this.sendMessage(1); + return "kafka"; + } + + public void sendMessage(Integer id) throws ExecutionException, InterruptedException { + producer.syncSend(id); + } + +} diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/message/DemoMessage.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/message/DemoMessage.java new file mode 100644 index 00000000..90050c82 --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/message/DemoMessage.java @@ -0,0 +1,31 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo.message; + +/** + * 示例 Message 消息 + */ +public class DemoMessage { + + public static final String TOPIC = "DEMO"; + + /** + * 编号 + */ + private Integer id; + + public DemoMessage setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "DemoMessage{" + + "id=" + id + + '}'; + } + +} diff --git a/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/producer/DemoProducer.java b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/producer/DemoProducer.java new file mode 100644 index 00000000..2d49cccf --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/java/cn/iocoder/springboot/lab40/skywalkingdemo/producer/DemoProducer.java @@ -0,0 +1,25 @@ +package cn.iocoder.springboot.lab40.skywalkingdemo.producer; + +import cn.iocoder.springboot.lab40.skywalkingdemo.message.DemoMessage; +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 DemoProducer { + + @Resource + private KafkaTemplate kafkaTemplate; + + public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException { + // 创建 DemoMessage 消息 + DemoMessage message = new DemoMessage(); + message.setId(id); + // 同步发送消息 + return kafkaTemplate.send(DemoMessage.TOPIC, message).get(); + } + +} diff --git a/lab-40/lab-40-kafka/src/main/resources/application.yaml b/lab-40/lab-40-kafka/src/main/resources/application.yaml new file mode 100644 index 00000000..8a228066 --- /dev/null +++ b/lab-40/lab-40-kafka/src/main/resources/application.yaml @@ -0,0 +1,26 @@ +spring: + application: + name: demo-application-kafka + + # Kafka 配置项，对应 KafkaProperties 配置类 + kafka: + bootstrap-servers: 127.0.0.1:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔 + # Kafka Producer 配置项 + producer: + acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。 + retries: 3 # 发送失败时，重试发送的次数 + key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化 + value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化 + # Kafka Consumer 配置项 + consumer: + auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解 + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer + properties: + spring: + json: + trusted: + packages: cn.iocoder.springboot.lab40.skywalkingdemo.message # 消息 POJO 可信目录，解决 JSON 无法反序列化的问题 + # Kafka Consumer Listener 监听器配置 + listener: + missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错 diff --git a/lab-40/pom.xml b/lab-40/pom.xml index 76a36bf1..1c22cd34 100644 --- a/lab-40/pom.xml +++ b/lab-40/pom.xml @@ -15,9 +15,9 @@ lab-40-demo lab-40-springmvc lab-40-mysql - lab-40-redis + lab-40-kafka lab-40-logback