mirror of
https://gitee.com/yudaocode/SpringBoot-Labs.git
synced 2026-09-03 05:53:54 +08:00
增加 zipkin 示例（kafka）
This commit is contained in:
YunaiV
68
lab-40/lab-40-kafka/pom.xml
Normal file
68
lab-40/lab-40-kafka/pom.xml
Normal file
@@ -0,0 +1,68 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>2.1.11.RELEASE</version>
|
||||
<relativePath/> <!-- lookup parent from repository -->
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>lab-40-kafka</artifactId>
|
||||
|
||||
<dependencies>
|
||||
<!-- 引入 Spring-Kafka 依赖 -->
|
||||
<!-- 已经内置 kafka-clients 依赖，所以无需重复引入 -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.kafka</groupId>
|
||||
<artifactId>spring-kafka</artifactId>
|
||||
<version>2.2.11.RELEASE</version>
|
||||
</dependency>
|
||||
|
||||
<!-- 实现对 SpringMVC 的自动化配置 -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Brave 核心库 -->
|
||||
<!-- The below are needed to report traces to http://localhost:9411/api/v2/spans -->
|
||||
<dependency>
|
||||
<groupId>io.zipkin.brave</groupId>
|
||||
<artifactId>brave</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.zipkin.reporter2</groupId>
|
||||
<artifactId>zipkin-sender-okhttp3</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Adds the MVC class and method names to server spans -->
|
||||
<!-- Brave 对 Spring MVC 的支持 -->
|
||||
<dependency>
|
||||
<groupId>io.zipkin.brave</groupId>
|
||||
<artifactId>brave-instrumentation-spring-webmvc</artifactId>
|
||||
</dependency>
|
||||
<!-- Brave 对 Kafka 的支持 -->
|
||||
<dependency>
|
||||
<groupId>io.zipkin.brave</groupId>
|
||||
<artifactId>brave-instrumentation-kafka-clients</artifactId>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<dependencyManagement>
|
||||
<!-- Brave Bom 文件 -->
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>io.zipkin.brave</groupId>
|
||||
<artifactId>brave-bom</artifactId>
|
||||
<version>5.9.1</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
</project>
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Span> 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);
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 +
|
||||
'}';
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Object, Object> kafkaTemplate;
|
||||
|
||||
public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
|
||||
// 创建 DemoMessage 消息
|
||||
DemoMessage message = new DemoMessage();
|
||||
message.setId(id);
|
||||
// 同步发送消息
|
||||
return kafkaTemplate.send(DemoMessage.TOPIC, message).get();
|
||||
}
|
||||
|
||||
}
|
||||
26
lab-40/lab-40-kafka/src/main/resources/application.yaml
Normal file
26
lab-40/lab-40-kafka/src/main/resources/application.yaml
Normal file
@@ -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 ，解决报错
|
||||
@@ -15,9 +15,9 @@
|
||||
<module>lab-40-demo</module>
|
||||
<module>lab-40-springmvc</module>
|
||||
<module>lab-40-mysql</module>
|
||||
|
||||
<module>lab-40-redis</module>
|
||||
|
||||
<module>lab-40-kafka</module>
|
||||
<module>lab-40-logback</module>
|
||||
</modules>
|
||||
|
||||
|
||||
Reference in New Issue
Block a user