Sleuth+Zipkin链路追踪

一：链路追踪

1.什么是链路追踪
　“链路追踪”一词是在 2010 年提出的，当时谷歌发布了一篇 Dapper 论文：Dapper，大规模分布式
系统的跟踪系统，介绍了谷歌自研的分布式链路追踪的实现原理，还介绍了他们是怎么低成本实现对应
用透明的。
　单纯的理解链路追踪，就是指一次任务的开始到结束，期间调用的所有系统及耗时（时间跨度）都
可以完整记录下来。
　　其实 Dapper 一开始只是一个独立的调用链路追踪系统，后来逐渐演化成了监控平台，并且基于监
控平台孕育出了很多工具，比如实时预警、过载保护、指标数据查询等。
　　除了谷歌的 Dapper，还有一些其他比较有名的产品，比如阿里的鹰眼、大众点评的 CAT、Twitter
的 Zipkin、Naver（著名社交软件LINE的母公司）的 PinPoint 以及国产开源的 SkyWalking（已贡献给
Apache） 等。

二：Sleuth介绍

1.什么是Sleuth
spring Cloud Sleuth 为 Spring Cloud 实现了分布式跟踪解决方案。兼容 Zipkin，HTrace 和其他基
于日志的追踪系统，例如 ELK（Elasticsearch 、Logstash、 Kibana）。

Spring Cloud Sleuth 提供了以下功能：
链路追踪 ：通过 Sleuth 可以很清楚的看出一个请求都经过了那些服务，可以很方便的理清服务间
的调用关系等。
性能分析 ：通过 Sleuth 可以很方便的看出每个采样请求的耗时，分析哪些服务调用比较耗时，当
服务调用的耗时随着请求量的增大而增大时， 可以对服务的扩容提供一定的提醒。
数据分析，优化链路 ：对于频繁调用一个服务，或并行调用等，可以针对业务做一些优化措施。
可视化错误 ：对于程序未捕获的异常，可以配合 Zipkin 查看。

2.span
基本工作单位，一次单独的调用链可以称为一个 Span，Dapper 记录的是 Span 的名称，以及每个
Span 的 ID 和父 ID，以重建在一次追踪过程中不同 Span 之间的关系，图中一个矩形框就是一个
Span，前端从发出请求到收到回复就是一个 Span。

开始跟踪的初始跨度称为 root span 。该跨度的 ID 的值等于跟踪 ID。
Dapper 记录了 span 名称，以及每个 span 的 ID 和父 span ID，以重建在一次追踪过程中不同
span 之间的关系。如果一个 span 没有父 ID 被称为 root span。所有 span 都挂在一个特定的 Trace
上，也共用一个 trace id。

3.Trace
一系列 Span 组成的树状结构，一个 Trace 认为是一次完整的链路，内部包含 n 多个 Span。Trace
和 Span 存在一对多的关系，Span 与 Span 之间存在父子关系。
　　举个例子：客户端调用服务 A 、服务 B 、服务 C 、服务 F，而每个服务例如 C 就是一个 Span，如
果在服务 C 中另起线程调用了 D，那么 D 就是 C 的子 Span，如果在服务 D 中另起线程调用了 E，那么
E 就是 D 的子 Span，这个 C -> D -> E 的链路就是一条 Trace。

4.Annotation
用来及时记录一个事件的存在，一些核心 annotations 用来定义一个请求的开始和结束。
cs - Client Sent：客户端发起一个请求，这个 annotation 描述了这个 span 的开始；
sr - Server Received：服务端获得请求并准备开始处理它，如果 sr 减去 cs 时间戳便可得到网络延
迟；
ss - Server Sent：请求处理完成（当请求返回客户端），如果 ss 减去 sr 时间戳便可得到服务端处
理请求需要的时间；
cr - Client Received：表示 span 结束，客户端成功接收到服务端的回复，如果 cr 减去 cs 时间戳
便可得到客户端从服务端获取回复的所有所需时间。

5.实现原理
　 如果想知道一个接口在哪个环节出现了问题，就必须清楚该接口调用了哪些服务，以及调用的顺
序，如果把这些服务串起来，看起来就像链条一样，我们称其为调用链。

想要实现调用链，就要为每次调用做个标识，然后将服务按标识大小排列，可以更清晰地看出调用
顺序，我们暂且将该标识命名为 spanid。

三：链路追踪案例：

3.1环境准备

　　
eureka-server ：注册中心
eureka-server02 ：注册中心
gateway-server ：Spring Cloud Gateway 服务网关
product-service ：商品服务，提供了根据主键查询商品接口
http://localhost:7070/product/{id} 根据多个主键查询商品接口
http://localhost:7070/product/listByIds
order-service ：订单服务，提供了根据主键查询订单接口
http://localhost:9090/order/{id} 且订单服务调用商品服务。
1
2
3
4
5
6
7
8
9

3.2.添加依赖

需要进行链路追踪的项目中（服务网关、商品服务、订单服务）添加 spring-cloud-startersleuth 依赖。

dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-sleuth</artifactId>
</dependency>
1
2
3
4

3.3.记录日志

在需要链路追踪的项目中（resource路径下手动添加个logback.xml）添加 logback.xml 日志文件，内容如下（logback 日志的输出级别需要是 DEBUG 级别

//注意修改 中项目名称。
//日志核心配置： %d{yyyy-MM-dd HH:mm:ss.SSS} [${applicationName},%X{X-B3-TraceId:-},%X{X-B3-SpanId:-}] [%thread] %-5level %logger{50} - %msg%n

<?xml version="1.0" encoding="UTF-8"?>
<!-- 日志级别从低到高分为TRACE < DEBUG < INFO < WARN < ERROR < FATAL，如果设置为WARN，
则低于WARN的信息都不会输出 -->
<!-- scan: 当此属性设置为true时，配置文件如果发生改变，将会被重新加载，默认值为true -->
<!-- scanPeriod: 设置监测配置文件是否有修改的时间间隔，如果没有给出时间单位，默认单位是毫秒。
当scan为true时，此属性生效。默认的时间间隔为1分钟。 -->
<!-- debug: 当此属性设置为true时，将打印出logback内部日志信息，实时查看logback运行状态。默
认值为false。 -->
<configuration scan="true" scanPeriod="10 seconds">
    <!-- 日志上下文名称 -->
    <contextName>my_logback</contextName>
    <!-- name的值是变量的名称，value的值是变量定义的值。通过定义的值会被插入到logger上下文
中。定义变量后，可以使“${}”来使用变量。 -->
    <property name="log.path" value="${catalina.base}/gateway-server/logs"/>
    <!-- 加载 Spring 配置文件信息 -->
    <springProperty scope="context" name="applicationName"
source="spring.application.name" defaultValue="localhost"/>
    <!-- 日志输出格式 -->
    <property name="LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} 
[${applicationName},%X{X-B3-TraceId:-},%X{X-B3-SpanId:-}] [%thread] %-5level 
%logger{50} - %msg%n"/>
    <!--输出到控制台-->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <!--此日志appender是为开发使用，只配置最底级别，控制台输出的日志级别是大于或等于此级
别的日志信息-->
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>DEBUG</level>
        </filter>
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
            <!-- 设置字符集 -->
            <charset>UTF-8</charset>
        </encoder>
    </appender>
    <!-- 输出到文件 -->
    <!-- 时间滚动输出 level为 DEBUG 日志 -->
    <appender name="DEBUG_FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 正在记录的日志文件的路径及文件名 -->
        <file>${log.path}/log_debug.log</file>
        <!--日志文件输出格式-->
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
            <charset>UTF-8</charset> <!-- 设置字符集 -->
        </encoder>
        <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
        <rollingPolicy
class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!-- 日志归档 -->
            <fileNamePattern>${log.path}/debug/log-debug-%d{yyyy-MMdd}.%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <!--日志文件保留天数-->
            <maxHistory>15</maxHistory>
        </rollingPolicy>
        <!-- 此日志文件只记录debug级别的 -->
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>DEBUG</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <!-- 时间滚动输出 level为 INFO 日志 -->
    <appender name="INFO_FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 正在记录的日志文件的路径及文件名 -->
        <file>${log.path}/log_info.log</file>
        <!--日志文件输出格式-->
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
            <charset>UTF-8</charset>
        </encoder>
        <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
        <rollingPolicy
class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!-- 每天日志归档路径以及格式 -->
            <fileNamePattern>${log.path}/info/log-info-%d{yyyy-MMdd}.%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <!--日志文件保留天数-->
            <maxHistory>15</maxHistory>
        </rollingPolicy>
        <!-- 此日志文件只记录info级别的 -->
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>INFO</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <!-- 时间滚动输出 level为 WARN 日志 -->
    <appender name="WARN_FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 正在记录的日志文件的路径及文件名 -->
        <file>${log.path}/log_warn.log</file>
        <!--日志文件输出格式-->
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
            <charset>UTF-8</charset> <!-- 此处设置字符集 -->
        </encoder>
        <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
        <rollingPolicy
class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${log.path}/warn/log-warn-%d{yyyy-MMdd}.%i.log</fileNamePattern>
            <!-- 每个日志文件最大100MB -->
            <timeBasedFileNamingAndTriggeringPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <!--日志文件保留天数-->
            <maxHistory>15</maxHistory>
        </rollingPolicy>
        <!-- 此日志文件只记录warn级别的 -->
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>WARN</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <!-- 时间滚动输出 level为 ERROR 日志 -->
    <appender name="ERROR_FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 正在记录的日志文件的路径及文件名 -->
        <file>${log.path}/log_error.log</file>
        <!--日志文件输出格式-->
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
            <charset>UTF-8</charset> <!-- 此处设置字符集 -->
        </encoder>
        <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
        <rollingPolicy
class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${log.path}/error/log-error-%d{yyyy-MMdd}.%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <!--日志文件保留天数-->
            <maxHistory>15</maxHistory>
            <!-- 日志量最大 10 GB -->
            <totalSizeCap>10GB</totalSizeCap>
        </rollingPolicy>
        <!-- 此日志文件只记录ERROR级别的 -->
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>ERROR</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <!-- 对于类路径以 com.example.logback 开头的Logger,输出级别设置为warn,并且只输出到控
制台 -->
    <!-- 这个logger没有指定appender，它会继承root节点中定义的那些appender -->
    <!-- <logger name="com.example.logback" level="warn"/> -->
    <!--通过 LoggerFactory.getLogger("myLog") 可以获取到这个logger-->
    <!--由于这个logger自动继承了root的appender，root中已经有stdout的appender了，自己这边
又引入了stdout的appender-->
    <!--如果没有设置 additivity="false" ,就会导致一条日志在控制台输出两次的情况-->
    <!--additivity表示要不要使用rootLogger配置的appender进行输出-->
    <logger name="myLog" level="INFO" additivity="false">
        <appender-ref ref="CONSOLE"/>
    </logger>
    <!-- 日志输出级别及方式 -->
    <root level="DEBUG">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="DEBUG_FILE"/>
        <appender-ref ref="INFO_FILE"/>
        <appender-ref ref="WARN_FILE"/>
        <appender-ref ref="ERROR_FILE"/>
    </root>
</configuration>
　　
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

3.4.访问（这里只是举例）

访问：http://localhost:9000/order-service/order/1 ，结果如下：
　　服务网关打印信息：[gateway-server,95aa725089b757f8,95aa725089b757f8]
　　商品服务打印信息 [product-service,95aa725089b757f8,e494e064842ce4e8]
　　订单服务打印信息 [product-service,95aa725089b757f8,e494e064842ce4e8]
　　通过打印信息可以得知，整个链路的 traceId 为： 95aa725089b757f8 ， spanId 为：
e494e064842ce4e8 和 f4ee41a6dcf08717 。

四：sleuth结合Zipkin进行链路追踪

4.1.什么是 Zipkin

Zipkin 是 Twitter 公司开发贡献的一款开源的分布式实时数据追踪系统（Distributed Tracking
System），基于 Google Dapper 的论文设计而来，其主要功能是聚集各个异构系统的实时监控数据。
　　它可以收集各个服务器上请求链路的跟踪数据，并通过 Rest API 接口来辅助我们查询跟踪数据，实
现对分布式系统的实时监控，及时发现系统中出现的延迟升高问题并找出系统性能瓶颈的根源。除了面
向开发的 API 接口之外，它还提供了方便的 UI 组件，每个服务向 Zipkin 报告计时数据，Zipkin 会根据
调用关系生成依赖关系图，帮助我们直观的搜索跟踪信息和分析请求链路明细。Zipkin 提供了可插拔数
据存储方式：In-Memory、MySql、Cassandra 以及 Elasticsearch。
　　分布式跟踪系统还有其他比较成熟的实现，例如：Naver 的 PinPoint、Apache 的 HTrace、阿里的
鹰眼 Tracing、京东的 Hydra、新浪的 Watchman，美团点评的 CAT，Apache 的 SkyWalking 等。

4.2.工作原理

共有四个组件构成了 Zipkin：
Collector ：收集器组件，处理从外部系统发送过来的跟踪信息，将这些信息转换为 Zipkin 内部
处理的 Span 格式，以支持后续的存储、分析、展示等功能。
Storage ：存储组件，处理收集器接收到的跟踪信息，默认将信息存储在内存中，可以修改存储策
略使用其他存储组件，支持 MySQL，Elasticsearch 等。
Web UI ：UI 组件，基于 API 组件实现的上层应用，提供 Web 页面，用来展示 Zipkin 中的调用链
和系统依赖关系等。
RESTful API ：API 组件，为 Web 界面提供查询存储中数据的接口。
　　
　　Zipkin 分为两端，一个是 Zipkin 服务端，一个是 Zipkin 客户端，客户端也就是微服务的应用，客
户端会配置服务端的 URL 地址，一旦发生服务间的调用的时候，会被配置在微服务里面的 Sleuth 的监
听器监听，并生成相应的 Trace 和 Span 信息发送给服务端。发送的方式有两种，一种是消息总线的方
式如 RabbitMQ 发送，还有一种是 HTTP 报文的方式发

4.3.服务端部署

服务端是一个独立的可执行的 jar 包，官方下载地址：https://search.maven.org/remote_conten
t?g=io.zipkin&a=zipkin-server&v=LATEST&c=exec，使用 java -jar zipkin.jar 命令启动，端口默
认为 9411 。我们下载的 jar 包为：zipkin-server-2.20.1-exec.jar，启动命令如下：

java -jar zipkin-server-2.20.1-exec.jar
//　访问：http://localhost:9411/ 即可验证是否启动成功
1
2

4.4.服务端部署

需要进行链路追踪的项目中（服务网关、商品服务、订单服务）添加 spring-cloud-starterzipkin 依赖。

<!-- spring cloud zipkin 依赖 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zipkin</artifactId>
</dependency>
1
2
3
4
5

配置文件
　　在需要进行链路追踪的项目中（服务网关、商品服务、订单服务）配置 Zipkin 服务端地址及数据传
输方式。默认即如下配置。
　
appclation.yaml

spring:
 zipkin:
   base-url: http://localhost:9411/ # 服务端地址
   sender:
     type: web                      # 数据传输方式，web 表示以 HTTP 报文的形式向服务端
发送数据
 sleuth:
   sampler:
     probability: 1.0               # 收集数据百分比，默认 0.1（10%）

1
2
3
4
5
6
7
8
9
10

4.5存储追踪数据

Zipkin Server 默认存储追踪数据至内存中，这种方式并不适合生产环境，一旦 Server 关闭重启或
者服务崩溃，就会导致历史数据消失。Zipkin 支持修改存储策略使用其他存储组件，支持 MySQL，
Elasticsearch 等。

4.6MySQL

打开 MySQL 数据库，创建 zipkin 库，执行以下 SQL 脚本

--
-- Copyright 2015-2019 The OpenZipkin Authors
--
-- Licensed under the Apache License, Version 2.0 (the "License"); you may not 
use this file except
-- in compliance with the License. You may obtain a copy of the License at
--
-- http://www.apache.org/licenses/LICENSE-2.0
--
-- Unless required by applicable law or agreed to in writing, software 
distributed under the License
-- is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 
KIND, either express
-- or implied. See the License for the specific language governing permissions 
and limitations under
-- the License.
--
CREATE TABLE IF NOT EXISTS zipkin_spans (
  `trace_id_high` BIGINT NOT NULL DEFAULT 0 COMMENT 'If non zero, this means the 
trace uses 128 bit traceIds instead of 64 bit',
  `trace_id` BIGINT NOT NULL,
  `id` BIGINT NOT NULL,
  `name` VARCHAR(255) NOT NULL,
  `remote_service_name` VARCHAR(255),
  `parent_id` BIGINT,
  `debug` BIT(1),
  `start_ts` BIGINT COMMENT 'Span.timestamp(): epoch micros used for endTs query 
and to implement TTL',
  `duration` BIGINT COMMENT 'Span.duration(): micros used for minDuration and 
maxDuration query',
  PRIMARY KEY (`trace_id_high`, `trace_id`, `id`)
) ENGINE=InnoDB ROW_FORMAT=COMPRESSED CHARACTER SET=utf8 COLLATE
utf8_general_ci;
ALTER TABLE zipkin_spans ADD INDEX(`trace_id_high`, `trace_id`) COMMENT 'for 
getTracesByIds';
ALTER TABLE zipkin_spans ADD INDEX(`name`) COMMENT 'for getTraces and 
getSpanNames';
ALTER TABLE zipkin_spans ADD INDEX(`remote_service_name`) COMMENT 'for getTraces 
and getRemoteServiceNames';
ALTER TABLE zipkin_spans ADD INDEX(`start_ts`) COMMENT 'for getTraces ordering 
and range';
CREATE TABLE IF NOT EXISTS zipkin_annotations (
  `trace_id_high` BIGINT NOT NULL DEFAULT 0 COMMENT 'If non zero, this means the 
trace uses 128 bit traceIds instead of 64 bit',
  `trace_id` BIGINT NOT NULL COMMENT 'coincides with zipkin_spans.trace_id',
  `span_id` BIGINT NOT NULL COMMENT 'coincides with zipkin_spans.id',
  `a_key` VARCHAR(255) NOT NULL COMMENT 'BinaryAnnotation.key or 
Annotation.value if type == -1',
  `a_value` BLOB COMMENT 'BinaryAnnotation.value(), which must be smaller than 
64KB',
  `a_type` INT NOT NULL COMMENT 'BinaryAnnotation.type() or -1 if Annotation',
  `a_timestamp` BIGINT COMMENT 'Used to implement TTL; Annotation.timestamp or 
zipkin_spans.timestamp',
  `endpoint_ipv4` INT COMMENT 'Null when Binary/Annotation.endpoint is null',
  `endpoint_ipv6` BINARY(16) COMMENT 'Null when Binary/Annotation.endpoint is 
null, or no IPv6 address',
  `endpoint_port` SMALLINT COMMENT 'Null when Binary/Annotation.endpoint is 
null',
  `endpoint_service_name` VARCHAR(255) COMMENT 'Null when 
Binary/Annotation.endpoint is null'
) ENGINE=InnoDB ROW_FORMAT=COMPRESSED CHARACTER SET=utf8 COLLATE
utf8_general_ci;
ALTER TABLE zipkin_annotations ADD UNIQUE KEY(`trace_id_high`, `trace_id`, 
`span_id`, `a_key`, `a_timestamp`) COMMENT 'Ignore insert on duplicate';
ALTER TABLE zipkin_annotations ADD INDEX(`trace_id_high`, `trace_id`, `span_id`)
COMMENT 'for joining with zipkin_spans';

-------------以上为SQL脚本
　　
部署 Zipkin 服务端
　　添加启动参数，重新部署服务端：
　　官网地址：https://github.com/openzipkin/zipkin/blob/master/zipkin-server/src/main/resource
s/zipkin-server-shared.yml
　　
测试
　　
　　访问：http://localhost:9000/order-service/order/1 查看数据库结果如下：
ALTER TABLE zipkin_annotations ADD INDEX(`trace_id_high`, `trace_id`) COMMENT
'for getTraces/ByIds';
ALTER TABLE zipkin_annotations ADD INDEX(`endpoint_service_name`) COMMENT 'for 
getTraces and getServiceNames';
ALTER TABLE zipkin_annotations ADD INDEX(`a_type`) COMMENT 'for getTraces and 
autocomplete values';
ALTER TABLE zipkin_annotations ADD INDEX(`a_key`) COMMENT 'for getTraces and 
autocomplete values';
ALTER TABLE zipkin_annotations ADD INDEX(`trace_id`, `span_id`, `a_key`) COMMENT
'for dependencies job';
CREATE TABLE IF NOT EXISTS zipkin_dependencies (
  `day` DATE NOT NULL,
  `parent` VARCHAR(255) NOT NULL,
  `child` VARCHAR(255) NOT NULL,
  `call_count` BIGINT,
  `error_count` BIGINT,
  PRIMARY KEY (`day`, `parent`, `child`)
) ENGINE=InnoDB ROW_FORMAT=COMPRESSED CHARACTER SET=utf8 COLLATE
utf8_general_ci;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

4.7部署zipKin服务端

添加启动参数，重新部署服务端：

java -jar zipkin-server-2.20.1-exec.jar --STORAGE_TYPE=mysql --
MYSQL_HOST=localhost --MYSQL_TCP_PORT=3306 --MYSQL_USER=root --MYSQL_PASS=root -
-MYSQL_DB=zipkin

1
2
3
4

4.8结合RabbitMq

1.安装启动RabbitMq服务端

systemctl start rabbitmq-server.service

//部署 Zipkin 服务端
　添加启动参数，重新部署服务端：
　java -jar zipkin-server-2.20.1-exec.jar --STORAGE_TYPE=mysql --
MYSQL_HOST=localhost --MYSQL_TCP_PORT=3306 --MYSQL_USER=root --MYSQL_PASS=root -
-MYSQL_DB=zipkin --RABBIT_ADDRESSES=192.168.10.101:5672 --RABBIT_USER=guest --
RABBIT_PASSWORD=guest --RABBIT_VIRTUAL_HOST=/ --RABBIT_QUEUE=zipkin
1
2
3
4
5
6
7
8

2.客户端添加依赖

<!-- spring cloud zipkin 依赖 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zipkin</artifactId>
</dependency>
<!-- 消息队列通用依赖 -->
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9
10

3.客户端配置文件

spring:
 zipkin:
   base-url: http://localhost:9411/ # 服务端地址
   sender:
     type: rabbit
   rabbitmq:
     queue: zipkin                  # 队列名称
 rabbitmq:
   host: 192.168.10.101             # 服务器 IP
   port: 5672                       # 服务器端口
   username: guest                  # 用户名
   password: guest                  # 密码
   virtual-host: /                  # 虚拟主机地址
   listener:
     direct:
       retry:
         enabled: true              # 是否开启发布重试
         max-attempts: 5            # 最大重试次数
         initial-interval: 5000     # 重试间隔时间（单位毫秒）
     simple:
       retry:
         enabled: true              # 是否开启消费者重试
         max-attempts: 5            # 最大重试次数
         initial-interval: 5000     # 重试间隔时间（单位毫秒）
 sleuth:
   sampler:
     probability: 1.0               # 收集数据百分比，默认 0.1（10%
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

五：结合ELK

1.搭建es集群（3个）
启动 head 插件，访问：http://192.168.10.101:9100/
2.部署Zpkin服务端

//　添加启动参数，重新部署服务端
java -jar zipkin-server-2.20.1-exec.jar --STORAGE_TYPE=elasticsearch --
ES_HOSTS=http://192.168.10.101:9200/,http://192.168.10.102:9200/,http://192.168.
10.103:9200/ --RABBIT_ADDRESSES=192.168.10.101:5672 --RABBIT_USER=guest --
RABBIT_PASSWORD=guest --RABBIT_QUEUE=zipkin
　//启动参数中包含 Elasticsearch 和 RabbitMQ 的配置，实现基于 MQ 并存储链路信息至
//Elasticsearch。

查看索引库
访问：http://192.168.10.101:9100 查看是否已经创建好了 zipkin 索引库。```

3.客户端添加依赖
```java
<!-- spring cloud zipkin 依赖 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zipkin</artifactId>
</dependency>
<!-- 消息队列通用依赖 -->
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

4、客户端配置文件

spring:
 zipkin:
   base-url: http://localhost:9411/ # 服务端地址
   sender:
     type: rabbit
   rabbitmq:
     queue: zipkin                  # 队列名称
 rabbitmq:
   host: 192.168.10.101             # 服务器 IP
   port: 5672                       # 服务器端口
   username: guest                  # 用户名
   password: guest                  # 密码
   virtual-host: /                  # 虚拟主机地址
   listener:
     direct:
       retry:
         enabled: true              # 是否开启发布重试
         max-attempts: 5            # 最大重试次数
         initial-interval: 5000     # 重试间隔时间（单位毫秒）
     simple:
       retry:
         enabled: true              # 是否开启消费者重试
         max-attempts: 5            # 最大重试次数
         initial-interval: 5000     # 重试间隔时间（单位毫秒）
 sleuth:
   sampler:
     probability: 1.0               # 收集数据百分比，默认 0.1（10%）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

5.ELK：
Elasticsearch 简称 ES：实时的分布式搜索和分析引擎，它可以用于全文搜索，结构化搜索以及分
析。建立在全文搜索引擎 Apache Lucene 基础上的搜索引擎，使用 Java 语言编写。
Logstash：具有实时传输能力的数据收集引擎，将各种各样的数据进行收集、解析，并发送给
ES。使用 Ruby 语言编写。
Kibana：为 Elasticsearch 提供了分析和可视化的 Web 平台。它可以在 Elasticsearch 的索引中查
找，交互数据，并生成各种维度表格、图形。
Beats：一组轻量级采集程序的统称，使用 Go 语言编写。以下是 elastic 官方支持的 5 种 beats，
事实上，伟大的开源力量早已创造出大大小小几十甚至上百种 beats，只有你没想到的，没有
beats 做不到的：
Filebeat：进行文件和目录采集，主要用于收集日志数据。
Winlogbeat：专门针对 Windows 的 event log 进行的数据采集。
Metricbeat：进行指标采集，指标可以是系统的，也可以是众多中间件产品的，主要用于监
控系统和软件的性能。
Packetbeat：通过网络抓包、协议分析，对一些请求响应式的系统通信进行监控和数据收
集，可以收集到很多常规方式无法收集到的信息。
Heartbeat：系统间连通性检测，比如 icmp，tcp，http 等系统的连通性监控
6.ELK环境准备

本文使用的 Elasticsearch 集群地址为：
192.168.10.101:9200
192.168.10.102:9200
192.168.10.103:9200
本文使用的 Logstash 的地址为：
192.168.10.101:9250
本文使用的 Kibana 的地址为：
192.168.10.101:5601
1
2
3
4
5
6
7
8

Logstash 运行时指定的配置文件 log-to-es.conf 内容如下：

# 数据入口
input {
 tcp {
 mode => "server"
 host => "192.168.10.101"
 port => 9250
 }
}
# 处理数据
 # 获取 @timestamp 的值并加上 8*60*60（北京时间比 logstash 中@timestamp 晚了 8 小
时），然后赋值给变量 timestamp。
filter {
 ruby { 
 code => "event.set('timestamp', event.get('@timestamp').time.localtime + 
8*60*60)" 
 }
 
 # 将 timestamp 值重新赋值给 @timestamp
 ruby {
 code => "event.set('@timestamp', event.get('timestamp'))"
 }
 # 删除变量 timestamp
 mutate {
 remove_field => ["timestamp"]
 }
}
# 数据出口
output {
 elasticsearch {
 hosts => ["192.168.10.101:9200", "192.168.10.102:9200", 
"192.168.10.103:9200"]
 index => "applog"
 }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

添加依赖
在需要进行链路追踪的项目中（服务网关、商品服务、订单服务）添加 logstash-logbackencoder 依赖。

<!-- logstash 编码依赖 -->
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>6.3</version>
</dependency>
1
2
3
4
5
6

日志配置
在需要进行链路追踪的项目中（服务网关、商品服务、订单服务）添加 logstash 输出 JSON 格式
数据 。

//logback.xml
<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="10 seconds">
    
   ...
    <!-- 为 Logstash 输出 JSON 格式数据 -->
    <appender name="LOGSTASH_PATTERN"
class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <!-- 数据输出目的地 -->
        <destination>192.168.10.101:9250</destination>
        <!-- 日志输出编码 -->
　　
查看索引库
　　
　　访问：http://192.168.10.101:9100 可以看到已经创建好了 applog 索引库。
        <encoder
class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <pattern>
                    <pattern>
                       {
                       "severity": "%level",
                       "service": "${springAppName:-}",
                       "trace": "%X{X-B3-TraceId:-}",
                       "span": "%X{X-B3-SpanId:-}",
                       "exportable": "%X{X-Span-Export:-}",
                       "pid": "${PID:-}",
                       "thread": "%thread",
                       "class": "%logger{40}",
                       "rest": "%message"
                       }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
    </appender>
    <!-- 日志输出级别及方式 -->
    <root level="DEBUG">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="LOGSTASH_PATTERN"/>
        <appender-ref ref="DEBUG_FILE"/>
        <appender-ref ref="INFO_FILE"/>
        <appender-ref ref="WARN_FILE"/>
        <appender-ref ref="ERROR_FILE"/>
    </root>
    
   ...
</configuration>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

访问：http://192.168.10.101:9100 查看是否已经创建好了 applog 索引库
访问：http://192.168.10.101:5601/ Kibana 首页。