你好,我是高楼。
在上一讲,我给你梳理了链路追踪的背景、目标、几种常见的组件,我们还确定了系统最后的选型方案。
在这一讲,我会通过案例演示在应用 Sleuth+Zipkin 来追踪我们电商微服务项目请求的过程中,需要关注的一些技术细节。
首先,我们来回顾一下电商项目集成 Sleuth+ Zipkin 的应用架构。
系统涉及的需要新增或改造的服务与组件包含以下 10 个:
结合技术栈,可以得出下面这张表格。表格整合了涉及追踪的技术组件,可以指导我们后续的改造工作。
接下来,我们通过 Demo 预演来一一做一下组件追踪的技术验证工作。
一般而言,我们使用 Brave 库,作为 Zipkin 客户端。同时它的 instrumentation 子项目,已经提供了 SpringMVC、Fegin 等组件的链路追踪功能。
所以通过引入 Spring Cloud Sleuth + Zipkin 相关依赖,可以实现对它们的自动配置,从而实现链路追踪的功能。
<!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId><version>2.2.6.RELEASE</version></dependency><!--Zipkin 客户端--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId><version>2.2.6.RELEASE</version></dependency>
从我给出的截图可以看出,已经成功引入 Zipkin、Sleuth 和 Brave 相关的依赖。
具体的示例,你可以参考这篇文章: Sleuth+Zipkin 实现 Spring Cloud 链路追踪 。
最后,Zipkin 可视化 UI 会出现对应的链路调用图及详细的链路。
因为 Brave 库默认提供了 Gateway 链路追踪的功能,所以要想实现 Sleuth 对 Spring Cloud Gateway 的代理请求的链路追踪,我们只需要集成就可以了。
这里我们结合上面示例中的三个服务,实现一个 API 网关,转发请求到 A 服务,即 Gateway -> A -> B -> C,各服务间通过 Fegin 实现远程调用。
注意,Spring Cloud Gateway 是基于 WebFlux 实现的,而 Spring Cloud Sleuth 的 instrument/web 模块提供的插件,实际是针对 WebFlux 框架,同样也适用于 Spring Cloud Gateway。
具体操作如下:
第一步,创建 pom.xml 文件,引入相关依赖。
<dependencies><!-- 引入 Spring Cloud Gateway 相关依赖,使用它作为网关,并实现对其的自动配置 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-gateway</artifactId></dependency><!-- 引入 Zipkin 依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency></dependencies>
第二步,在 application.yml 中,添加服务路由相关配置。
server:port: 8888spring:application:name: demo-gateway-applicationsleuth:sampler: #采样器probability: 1.0 #采样率,采样率是采集 Trace 的比率,默认 0.1rate: 10000 #每秒数据采集量,最多 n 条/秒 Traceweb: # Web 组件的配置项,例如说 SpringMVCenabled: truezipkin: #设置 zipkin 服务端地址base-url: http://127.0.0.1:9411cloud:# Spring Cloud Gateway 配置项,对应 GatewayProperties 类gateway:# 路由配置项,对应 RouteDefinition 数组routes:- id: feign-service-routeuri: http://127.0.0.1:7000predicates:- Path=/**
因为使用的是 instrument/web 模块提供的插件,所以和 SpringMVC 一样,WebFlux 也是使用 spring.sleuth.web 配置项。
配置项中,我们创建了一个编号为 feign-service-route 的路由,转发到 a-service 服务。
第三步,网关启动类。
package com.dunshan.gatewaydemo;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;/*** @author dunshan*/@SpringBootApplicationpublic class GatewayApplication {public static void main(String[] args) {SpringApplication.run(GatewayApplication.class, args);}}
最后启动所有应用,我们使用 Postman 测试一下接口,执行一次请求操作,尝试追踪该链路。
在 Zipkin 可视化 UI 就可以看到刚才我们调用接口的链路数据了。
这条链路经过 gateway、a-service、b-service、c-service 四个服务。
我们点开该链路,可以看到一个 Trace 明细。
这里需要实现一个简单的 MySQL 查询,接下来我会使用 MySQL8 驱动进行数据库的操作,这也是我们项目目前采用的方式。
目前 Brave 支持通过三种插件实现 MySQL 链路数据收集,它们分别是:
我简单演示一下用 Spring JDBC Template 的方式进行 MySQL 操作的步骤。
第一步,创建 pom.xml 文件,引入相关依赖。
<dependencies><!-- 引入 SpringMVC 相关依赖,并实现对其的自动配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 实现对数据库连接池的自动化配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency><!--Mysql 数据库驱动--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.15</version></dependency><!-- 引入 Zipkin 依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency><!-- Brave 对 MySQL8 的支持 --><dependency><groupId>io.zipkin.brave</groupId><artifactId>brave-instrumentation-mysql8</artifactId></dependency></dependencies>
这里引入 brave-instrumentation-mysql8 依赖,实现对 MySQL 的链路追踪。
第二步,在 application.yml 中,添加数据库相关配置。
spring:application:name: demo-service # 服务名sleuth:sampler: #采样器probability: 1.0 #采样率,采样率是采集 Trace 的比率,默认 0.1rate: 10000 #每秒数据采集量,最多 n 条/秒 Traceweb: # Web 组件的配置项,例如说 SpringMVCenabled: truezipkin: #设置 zipkin 服务端地址base-url: http://127.0.0.1:9411# datasource 数据源配置内容datasource:url: jdbc:mysql://127.0.0.1:3306/test?useSSL=false&useUnicode=true&characterEncoding=UTF-8&queryInterceptors=brave.mysql8.TracingQueryInterceptor&exceptionInterceptors=brave.mysql8.TracingExceptionInterceptor&zipkinServiceName=mysql-demodriver-class-name: com.mysql.jdbc.Driverusername: rootpassword: rootlogging:level:root: debug #为演示需要,开启 debug 级别日志
在这里,Brave 通过实现类 TracingQueryInterceptor 拦截 SQL 请求,进行 MySQL 的链路追踪。切记,在 spring.datasource.url 配置项上的 queryInterceptors、exceptionInterceptors 和 zipkinServiceName 属性上,一定要分别设置拦截器和该 MySQL 在 Zipkin 中展示的服务名。
第三步,在 MySQL 数据库中,创建用户表并插入数据。
CREATE TABLE `t_user` (`id` int(8) NOT NULL AUTO_INCREMENT COMMENT '主键自增',`username` varchar(50) NOT NULL COMMENT '用户名',`password` varchar(50) NOT NULL COMMENT '密码',PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='用户表';INSERT INTO `t_user`(`id`, `username`, `password`) VALUES (1, '7d', '123456');
第四步,创建 DemoController 类,提供示例 API 接口。
package com.dunshan.mysql8demo.controller;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.jdbc.core.BeanPropertyRowMapper;import org.springframework.jdbc.core.JdbcTemplate;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/7d")public class DemoController {@Autowiredprivate JdbcTemplate template;@GetMapping("/get")public String get(@RequestParam("id") Integer id) {this.selectById(1);return "success";}public Object selectById(Integer id) {return template.queryForObject("SELECT id, username, password FROM t_user WHERE id = ?",new BeanPropertyRowMapper<>(Object.class), // 结果转换成对应的对象。id);}}
第五步,创建 DemoServiceApplication 启动类。
package com.dunshan.mysql8demo;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class DemoServiceApplication {public static void main(String[] args) {SpringApplication.run(DemoServiceApplication.class, args);}}
这样在 /7d/get 接口会执行一次 MySQL 的查询操作。
最后启动应用,我们使用 Postman 测试一下接口,尝试追踪该链路。
同样在 Zipkin 可视化 UI 也可以看到刚才调用接口的链路数据。
点开这个链路,可以看到一个 Trace 明细。
到这里,我们的 MySQL 组件链路追踪就成功了。
这一部分需要实现简单的 Redis 操作,接下来我们使用 Spring Data Redis + Jedis 进行 Redis 的操作(项目部分服务使用的方式)。目前 Brave 暂未支持 Jedis 客户端的方式,所以我们只能考虑其它的办法。
在 opentracing-contrib 项目中,有一个 java-redis-client 子项目,提供了 OpenTracing 针对 Jedis、Lettuce 等客户端的链路追踪功能。这样,我们搭配上 brave-opentracing 项目,可以使用 OpenTracing API 收集的链路数据,发送给 Zipkin。
而 Lettuce 基于 Brave 实现了 BraveTracing ,从而可以实现对 Redis 操作的链路追踪。并且,Spring Cloud Sleuth 的 instrument/redis 模块对它实现了自动配置。如果项目中是使用 Lettuce 作为 Redis 客户端的话,可以考虑采用这种方式。
我们还是来看下具体的操作步骤。
第一步,创建 pom.xml 文件,引入相关依赖。
<dependencies><!-- 引入 SpringMVC 相关依赖,并实现对其的自动配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 实现对 Spring Data Redis 的自动化配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><exclusions><!-- 去掉对 Lettuce 的依赖,因为 Spring Boot 优先使用 Lettuce 作为 Redis 客户端 --><exclusion><groupId>io.lettuce</groupId><artifactId>lettuce-core</artifactId></exclusion></exclusions></dependency><!-- 引入 Jedis 的依赖 --><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId></dependency><!-- 引入 Zipkin 相关依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--引入 Sleuth 相关依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency><!-- Brave 对 Opentracing 的实现 --><dependency><groupId>io.opentracing.brave</groupId><artifactId>brave-opentracing</artifactId><version>0.35.0</version></dependency><!-- Opentracing 对 Redis 的支持 --><dependency><groupId>io.opentracing.contrib</groupId><artifactId>opentracing-redis-jedis3</artifactId><version>0.1.16</version></dependency><dependency><groupId>io.opentracing.contrib</groupId><artifactId>opentracing-redis-spring-data</artifactId><version>0.1.16</version></dependency></dependencies>
这里引入 brave-opentracing 依赖,也就是 Brave 对 Opentracing 的实现。
注意,Opentracing 和 JDBC 一样是一个通用标准,因此需要有 Brave 对 Opentracing 做具体实现,从而将链路数据写入到 Zipkin 中。就好比 JDBC 对 MySQL Driver 实现,将数据写入到 MySQL 数据库中一样。
同时,我们要引入 opentracing-redis-spring-data 和 opentracing-redis-jedis3 依赖,实现对 Jedis 操作 Redis 的链路追踪。
第二步,在 application.yml 中,添加redis相关配置。
spring:application:name: demo-service # 服务名sleuth:sampler: #采样器probability: 1.0 #采样率,采样率是采集 Trace 的比率,默认 0.1rate: 10000 #每秒数据采集量,最多 n 条/秒 Tracezipkin: #设置 zipkin 服务端地址base-url: http://127.0.0.1:9411# 对应 RedisProperties 类redis:host: 127.0.0.1port: 6379password: # Redis redis密码,默认为空。database: 0 # Redis redis中的数据库号,默认为 0。timeout: 6000 # Redis 连接超时时间,单位:毫秒。# 对应 RedisProperties.Jedis 内部类jedis:pool:max-active: 8 # 连接池最大连接数,默认为 8。使用负数表示没有限制。max-idle: 8 # 默认连接数最小空闲的连接数,默认为 8。使用负数表示没有限制。min-idle: 0 # 默认连接池最小空闲的连接数,默认为 0。允许设置 0 和 正数。max-wait: -1 # 连接池最大阻塞等待时间,单位:毫秒。默认为 -1,表示不限制。logging:level:root: debug #为演示需要,开启 debug 级别日志
第三步,创建 SleuthConfiguration 配置类,创建一个 TracingRedisConnectionFactory Bean 对象。这样,我们就能拦截到 Redis 操作,进行相应的链路跟踪了。
package com.dunsan.redisdemo.config;import io.opentracing.Tracer;import io.opentracing.contrib.redis.common.TracingConfiguration;import io.opentracing.contrib.redis.spring.data.connection.TracingRedisConnectionFactory;import org.springframework.boot.autoconfigure.data.redis.RedisProperties;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.data.redis.connection.RedisConnectionFactory;import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;@Configurationpublic class SleuthConfiguration {@Beanpublic RedisConnectionFactory redisConnectionFactory(Tracer tracer, RedisProperties redisProperties) {// 创建 JedisConnectionFactory 对象RedisConnectionFactory connectionFactory = new JedisConnectionFactory();// 创建 TracingConfiguration 对象TracingConfiguration tracingConfiguration = new TracingConfiguration.Builder(tracer)// 设置拓展 Tag,设置 Redis 服务器地址。因为默认情况下,不会在操作 Redis 链路的 Span 上记录 Redis 服务器的地址,所以这里需要设置。.extensionTag("Server Address", redisProperties.getHost() + ":" + redisProperties.getPort()).build();// 创建 TracingRedisConnectionFactory 对象return new TracingRedisConnectionFactory(connectionFactory, tracingConfiguration);}}
第四步,创建 DemoController 类,提供示例 API 接口。
package com.dunsan.redisdemo.controller;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/7d")public class DemoController {@Autowiredprivate StringRedisTemplate redisTemplate;@GetMapping("/get")public String get(@RequestParam("id") Integer id) {this.get("demo");return "success";}public void get(String key) {redisTemplate.opsForValue().get(key);}}
这样在 /7d/get 接口中,就会执行一次 Redis 的查询。
第五步,创建 DemoServiceApplication 启动类。
package com.dunsan.redisdemo;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class DemoServiceApplication {public static void main(String[] args) {SpringApplication.run(DemoServiceApplication.class, args);}}
最后启动应用。我们用 Postman 测试一下口,执行一次 Redis 查询操作,尝试跟踪该链路。
在 Zipkin 可视化 UI 中就可以看到刚才我们调用接口的链路数据了。
点开该链路,可以看到一个 Trace 明细。
到这里,我们的 Redis 组件链路跟踪也已经成功了。
这里还是实现简单的 MongoDB 操作,我会使用 Spring Data MongoDB + MongoTemplate 进行 MongoDB 的操作。目前,Brave 默认提供了对 MongoDB 操作的链路跟踪,通过 brave-instrumentation-mongo 库实现收集链路数据。
我们来看下具体的实现路径:
第一步,还是引入相关依赖。
<dependencies><!-- 引入 SpringMVC 相关依赖,并实现对其的自动配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 自动化配置 Spring Data Mongodb --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-mongodb</artifactId></dependency><!-- 引入 Zipkin 相关依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency><!--添加 brave mongodb 依赖 --><dependency><groupId>io.zipkin.brave</groupId><artifactId>brave-instrumentation-mongodb</artifactId><version>5.13.3</version></dependency>
第二步,创建全局配置文件,添加MongoDB配置。
spring:application:name: demo-service # 服务名sleuth:sampler: #采样器probability: 1.0 #采样率,采样率是采集 Trace 的比率,默认 0.1rate: 10000 #每秒数据采集量,最多 n 条/秒 Traceweb: # Web 组件的配置项,例如说 SpringMVCenabled: truezipkin: #设置 zipkin 服务端地址base-url: http://127.0.0.1:9411data:# MongoDB 配置项,对应 MongoProperties 类mongodb:host: 127.0.0.1port: 27017database: demo# username:# password:logging:level:root: debug #为演示需要,开启 debug 级别日志
第三步,创建 SleuthConfiguration 配置类,创建一个 TraceMongoCommandListener Bean 对象。这样,我们就能拦截到 MongoDB 操作,进行相应的链路跟踪了。
package com.dunshan.mongodbdemo.config;import brave.Tracer;import brave.Tracing;import brave.mongodb.MongoDBTracing;import com.mongodb.MongoClientSettings;import com.mongodb.client.MongoClient;import com.mongodb.client.MongoClients;import com.mongodb.event.CommandListener;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class SleuthConfiguration {@Beanpublic MongoClient mongoClient(Tracer tracer) {CommandListener listener = MongoDBTracing.create(Tracing.current()).commandListener();MongoClientSettings settings = MongoClientSettings.builder().addCommandListener(listener).build();MongoClient client = MongoClients.create(settings);return client;}}
第四步,创建 DemoController 类,提供示例 API 接口。
package com.dunshan.mongodbdemo.controller;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.mongodb.core.MongoTemplate;import org.springframework.data.mongodb.core.query.Criteria;import org.springframework.data.mongodb.core.query.Query;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;@RestController@RequestMapping("/7d")public class DemoController {@Autowiredprivate MongoTemplate mongoTemplate;@GetMapping("/get")public String get(@RequestParam("id") Integer id) {this.findById(1);return "success";}public DemoDO findById(Integer id) {return mongoTemplate.findOne(new Query(Criteria.where("_id").is(id)), DemoDO.class);}}
我们可以请求 /7d/get 接口,执行一次 MongoDB 的查询。
第五步,创建 DemoServiceApplication 启动类。
package com.dunshan.mongodbdemo;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class DemoServiceApplication {public static void main(String[] args) {SpringApplication.run(DemoServiceApplication.class, args);}}
最后启动应用。我们使用 Postman 测试接口,执行一次 MongoDB 查询操作,尝试跟踪该链路。
在 Zipkin 可视化 UI 中可以看到刚才我们调用接口的链路数据。
点开该链路,可以看到一个 Trace 明细。
我们的 MongoDB 组件链路跟踪也已经成功了。
因为 Brave 库也默认提供了 RabbitMQ 链路追踪的功能,所以同样只需要集成就可以了。
顺便提下 ,Brave 主要通过两个插件实现链路数据收集,它们分别是:
搭建生产者示例
第一步,还是引入相关依赖。
<dependencies><!-- 引入 SpringMVC 相关依赖,并实现对其的自动配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 引入 Spring Cloud Stream RabbitMQ 相关依赖,将 RabbitMQ 作为消息队列,并实现对其的自动配置 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><!-- 引入 Zipkin 依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency></dependencies>
第二步,创建全局配置文件,添加RabbitMQ配置。
spring:application:name: demo-producercloud:# Spring Cloud Stream 配置项,对应 BindingServiceProperties 类stream:# Binder 配置项,对应 BinderProperties Mapbinders:demo:type: rabbit # 设置 Binder 的类型environment: # 设置 Binder 的环境配置# 如果是 RabbitMQ 类型的时候,则对应的是 RabbitProperties 类spring:rabbitmq:host: 127.0.0.1 # 服务地址port: 5672 # 服务端口username: guest # 服务账号password: guest # 服务密码# Binding 配置项,对应 BindingProperties Mapbindings:demo-output:destination: DEMO-TOPIC # 目的地,使用 RabbitMQ Exchangecontent-type: application/json # 内容格式binder: demo # 设置使用的 Binder 名字sleuth:sampler: #采样器probability: 1.0 #采样率,采样率是采集 Trace 的比率,默认 0.1rate: 10000 #每秒数据采集量,最多 n 条/秒 Tracemessaging:# Spring Cloud Sleuth 针对 RabbitMQ 组件的配置项rabbit:enabled: true # 是否开启remote-service-name: rabbitmq # 远程服务名,默认为 rabbitmqzipkin: #设置 zipkin 服务端地址base-url: http://127.0.0.1:9411server:port: 18080
第三步,创建 MySource 接口,声明名字为 Output Binding。
package com.dunshan.rabbitmqdemo.producerdemo.message;import org.springframework.cloud.stream.annotation.Output;import org.springframework.messaging.MessageChannel;public interface MySource {@Output("demo-output")MessageChannel demoOutput();}
第四步,创建 DemoMessage 类,示例 Message 消息。
package com.dunshan.rabbitmqdemo.producerdemo.message;/*** 示例 Message 消息*/public class DemoMessage {/*** 编号*/private Integer id;public DemoMessage setId(Integer id) {this.id = id;return this;}public Integer getId() {return id;}@Overridepublic String toString() {return "DemoMessage{" +"id=" + id +'}';}}
第五步,创建 DemoController 类,提供示例 API 接口。
package com.dunshan.rabbitmqdemo.producerdemo.controller;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.messaging.Message;import org.springframework.messaging.support.MessageBuilder;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import java.util.Random;@RestController@RequestMapping("/7d")public class DemoController {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate MySource mySource;@GetMapping("/send")public boolean send() {// 创建 MessageDemoMessage message = new DemoMessage().setId(new Random().nextInt());// 创建 Spring Message 对象Message<DemoMessage> springMessage = MessageBuilder.withPayload(message).build();// 发送消息return mySource.demoOutput().send(springMessage);}}
我们可以请求 /7d/send 接口,执行一次发送消费。
第六步,创建 DemoServiceApplication 启动类。
package com.dunshan.rabbitmqdemo.producerdemo;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.cloud.stream.annotation.EnableBinding;@SpringBootApplication@EnableBinding(MySource.class)public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class, args);}}
这样,我们的生产者示例就搭建完了。
搭建消费者示例
第一步,还是引入相关依赖。
<dependencies><!-- 引入 SpringMVC 相关依赖,并实现对其的自动配置 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 引入 Spring Cloud Stream RabbitMQ 相关依赖,将 RabbitMQ 作为消息队列,并实现对其的自动配置 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><!-- 引入 Zipkin 依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency></dependencies>
第二步,创建全局配置文件,添加相关数据库配置。
spring:application:name: demo-consumercloud:# Spring Cloud Stream 配置项,对应 BindingServiceProperties 类stream:# Binder 配置项,对应 BinderProperties Mapbinders:demo:type: rabbit # 设置 Binder 的类型environment: # 设置 Binder 的环境配置# 如果是 RabbitMQ 类型的时候,则对应的是 RabbitProperties 类spring:rabbitmq:host: 127.0.0.1 # RabbitMQ 服务的地址port: 5672 # RabbitMQ 服务的端口username: guest # 服务账号password: guest # 服务密码# Binding 配置项,对应 BindingProperties Mapbindings:demo01-input:destination: DEMO-TOPIC # 目的地。这里使用 RabbitMQ Exchangecontent-type: application/json # 内容格式。这里使用 JSONgroup: demo-consumer-group-DEMO-TOPIC # 消费者分组binder: demo # 设置使用的 Binder 名字sleuth:sampler: #采样器probability: 1.0 #采样率,采样率是采集 Trace 的比率,默认 0.1rate: 10000 #每秒数据采集量,最多 n 条/秒 Tracemessaging:# Spring Cloud Sleuth 针对 RabbitMQ 组件的配置项rabbit:enabled: true # 是否开启remote-service-name: rabbitmq # 远程服务名,默认为 rabbitmqzipkin: #设置 zipkin 服务端地址base-url: http://127.0.0.1:9411server:port: ${random.int[10000,19999]} # 随机端口,方便启动多个消费者
spring.sleuth.messaging.rabbit 是 Spring Cloud Sleuth 针对 RabbitMQ 组件的配置项,对应 SleuthMessagingProperties.Rabbit 类。
第三步,创建 MySink 接口,声明名字为 Intput Binding。
package com.dunshan.rabbitmqdemo.consumerdemo.listener;import org.springframework.cloud.stream.annotation.Input;import org.springframework.messaging.SubscribableChannel;public interface MySink {String DEMO_INPUT = "demo-input";@Input(DEMO_INPUT)SubscribableChannel demoInput();}
第四步,创建 DemoMessage 类,示例 Message 消息。
package com.dunshan.rabbitmqdemo.consumerdemo.message;/*** 示例 Message 消息*/public class DemoMessage {/*** 编号*/private Integer id;public DemoMessage setId(Integer id) {this.id = id;return this;}public Integer getId() {return id;}@Overridepublic String toString() {return "DemoMessage{" +"id=" + id +'}';}}
第五步,创建 DemoConsumer 类,消费消息。
package com.dunshan.rabbitmqdemo.consumerdemo.listener;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.cloud.stream.annotation.StreamListener;import org.springframework.messaging.handler.annotation.Payload;import org.springframework.stereotype.Component;@Componentpublic class DemoConsumer {private Logger logger = LoggerFactory.getLogger(getClass());@StreamListener(MySink.DEMO_INPUT)public void onMessage(@Payload DemoMessage message) {logger.info("[onMessage][线程编号:{} 消息内容:{}]", Thread.currentThread().getId(), message);}}
第六步,创建 DemoServiceApplication 启动类。
package com.dunshan.rabbitmqdemo.consumerdemo;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.cloud.stream.annotation.EnableBinding;@SpringBootApplication@EnableBinding(MySink.class)public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}}
最后,启动应用。我们使用 Postman 测试一下接口,发送一条消息,尝试跟踪该链路。
查看结果,在 Zipkin 已经可以看到刚才我们调用接口的链路数据了。
点开该链路可以看到一个 Trace 明细。
到这里,我们的 RabbitMQ 组件链路跟踪也成功了。
刚才,我对所有的技术组件都单独进行了 demo 预演。单就这个过程来说,改造的复杂度还是可以接受的。接下来我们就一起来看看,怎么让它们在真实项目中落地。
我这里主要介绍三个服务的改造过程(其他服务是类似的操作),它们分别是:
mall-gateway 主要使用的是 Spring Cloud Gateway 技术,我们这个项目的主要用途是路由匹配、请求统一校验认证和鉴权。
大致的执行流程可以参考下图。
通过上面的介绍,我们知道 Sleuth 原生就支持对 Spring Cloud Gateway 链路追踪,所以我们只需要集成就可以了。
我们在 pom.xml 文件引入相关依赖。
<dependencies><!-- 引入 Spring Cloud Gateway 相关依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-gateway</artifactId></dependency><!-- 引入 Zipkin 依赖--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency></dependencies>
mall-auth 是认证中心服务,主要结合 Oauth2 实现用 JWT 令牌存储信息、刷新令牌功能还有权限认证等工作,涉及的组件主要为 Redis (非 Jedis 客户端方式)和 MySQL。
所以在 pom.xml 文件,引入相关依赖即可。
<!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency><!--Zipkin 客户端--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!-- Brave 对 MySQL8 的支持 --><dependency><groupId>io.zipkin.brave</groupId><artifactId>brave-instrumentation-mysql8</artifactId></dependency>
启动应用后,我们使用 API 文档测试下接口,尝试跟踪该链路。
在 Zipkin 就可以看到刚才我们调用接口的链路数据了,它分别涉及了四个组件。
我们点开该链路,可以看到一个 Trace 明细。
同样,我们也可以查看拓扑关系。
mall-member 是我们的会员服务,这里以核心接口【会员登录】为例,梳理一下它涉及到的组件。
通过分析代码调用链,我们知道它主要涉及 MySQL、Redis、MongoDB 等组件。
所以我们只需要在 pom.xml 文件引入相关依赖即可。
<!--添加 Sleuth 依赖 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-sleuth</artifactId></dependency><!--Zipkin 客户端--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-zipkin</artifactId></dependency><!-- Brave 对 MySQL8 的支持 --><dependency><groupId>io.zipkin.brave</groupId><artifactId>brave-instrumentation-mysql8</artifactId></dependency><!--添加 brave mongodb 依赖 --><dependency><groupId>io.zipkin.brave</groupId><artifactId>brave-instrumentation-mongodb</artifactId><version>5.13.3</version></dependency><!-- Brave 对 Opentracing 的实现 --><dependency><groupId>io.opentracing.brave</groupId><artifactId>brave-opentracing</artifactId><version>0.35.0</version></dependency><!-- Opentracing 对 Redis 的支持 --><dependency><groupId>io.opentracing.contrib</groupId><artifactId>opentracing-redis-jedis3</artifactId><version>0.1.16</version></dependency><dependency><groupId>io.opentracing.contrib</groupId><artifactId>opentracing-redis-spring-data</artifactId><version>0.1.16</version></dependency>
同时,要在 config 下新增两个配置类。
BraveConfiguration 为 MongoDB 链路追踪配置类,OpentracingConfiguration 为 Redis 组件链路追踪配置类。
启动应用后,我们使用 API 文档测试【会员登录】接口,尝试追踪该链路。
在 Zipkin 中可以看到刚才我们调用接口的链路数据,它分别涉及了五个组件。
我们点开这条链路,可以看到 Trace 明细。
也可以查找依赖,看到拓扑关系。
接着我们使用 API 文档测试一下【添加品牌关注】接口,尝试追踪该链路。
在 Zipkin 可以看到刚才我们调用接口的链路数据,它分别涉及了三个组件。
我们点开这条链路,也可以看到详细的 Trace 明细。
改造完服务后,我们还有一个需求,那就是对 Zipkin 生产环境的功能增强。也就是说, 我们这里将链路日志推送到 Kafka,然后启动 Zipkin Server 聚合日志,监听 Kafka ,如果有新的消息则进行拉取存入到 ElasticSeach,最后再用 Zipkin UI 展示链路过程。
这里的改造大概分为两个部分:
第一步,我们需要在各个服务引入 Kafka 依赖。
<!-- 引入 Spring Cloud Stream Kafka--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId></dependency>
第二步,修改 bootstrap.yml 配置文件,增加 Kafka 系列配置。
spring:zipkin: #设置zipkin服务端地址sender:type: kafka #指定发送到kafka,还可以指定Rabbit、Webservice:name: ${spring.application.name} #Zipkin链路日志中收集的服务名称kafka:topic: zipkinkafka:bootstrap-servers: kafka:9092 #Kubernetes中Kakfa地址,当然也可以指定Kubernetes集群外的Kafk
第三步,Zipkin 安装时增加 Storage 参数,比如 Docker 安装可以参考我给出的命令。
docker run -it -d --restart=always -e KAFKA_BOOTSTRAP_SERVERS=192.168.3.58:9092 -e STORAGE_TYPE=elasticsearch -e ES_HOSTS=http://192.168.3.58:9200 -p 9411:9411 openzipkin/zipkin
第四步,在 Kafka Manager 验证一下 Topic 情况。
最后,我们使用 ES Head 客户端插件查询一下链路日志。
可以看到,链路追踪数据已经成功保存了。
好了,这节课的内容到这里就全部讲完了。
刚才,我们进行了 demo 预演和系统实战,演示了如何在微服务项目中集成 Sleuth + Zipkin 落地链路追踪,希望能够给你一些启发。
这里有几个重点我们再一起回顾一下:
链路追踪现如今已成为微服务架构性能监控的标配,如何去选型并结合系统本身的特点做到成功的改造是不容易的。希望你可以动手实践起来,因为只有实践才是检验真理的唯一标准。
在课程的最后,我还是照例给你留两道思考题:
欢迎你在留言区和我交流讨论,我们下节课见!