在多线程环境下，使用 HashMap 进行 put 操作时由于有 resize 存在，因此会有死锁隐患，为了避免这种bug的隐患，强烈建议使用 ConcurrentHashMap 代替 HashMap，为了对更深入的了解，本文将对 JDK1.7 和 JDK1.8 的不同实现进行分析

jcmd 是 JDK1.7 之后出的命令行工具，如果你是 JDK1.7 之后的项目，建议你用 jcmd 替换掉 jps。
你可以使用它来查看堆信息：jcmd pid GC.heap_dump
也可以用来查看当前所有的 VM 虚拟机：jcmd -l
以及还有当前 VM 虚拟机的参数信息：jcmd PID VM.flags
具体更多命令：jcmd help jcmd PID help

1 微服务架构产生的历史背景

1.1 研发成本挑战

1.1.1 代码重复率

1. 从技术架构角度看，传统垂直架构的特点是本地 API 接口调用，不存在业务的拆分和互相调用，使用到什么功能就本地开发，非常方便，不需要过度依赖于其它功能模块。
2. 跨地域、跨开发小组协调很困难。

1.1.2 需求变更困难

代码重复率变高之后，已有功能变更或者新需求加入都会非常困难，以充值缴费功能为例，不同的充值渠道开发了相同的限额保护功能，当限额保护功能发生变更之后，所有重复开发的限额保护功能都需要重新修改和测试，很容易出现修改不一致或者被遗漏，导致部分渠道充值功能正常：

1.2 运维成本高

1.2.1 代码维护困难

传统的业务流程是由一长串本地接口或者方法调用串联起来的，而且往往由一个负责开发和维护。随着业务的发展和需求变化，本地diamante在不断地迭代和变更，最后形成了一个个垂直的功能孤岛，只有原来的开发者才理解接口调用关系和功能需求，一旦原来的开发者离职或调到其它项目组，这些功能模块的运维就会变得非常困难：

1.2.2 部署效率低

1. 业务没有拆分，很多功能模块都打到同一个 war 包中，一旦有一个功能发生变更，就需要重新打包和部署。
2. 测试工作量较大，因此存在大量重复的功能类库，需要针对所有调用方进行测试，测试工作量大。

1.3 新需求上线周期长

1. 新功能通常无法独立编译、打包、部署和上线，它可能混杂在老的系统中开发，很难剥离出来，这就无法通过服务灰度发布的形式快速上线。
2. 由于业务没有进行水平和垂直拆分，导致代码重复率高，新需求的开发、测试、打包和部署成本都比较高。

2 微服务架构带来的改变

2.1 应用解耦

微服务化之前，一个大型的应用系统通常会包含多个子应用，不同应用之间存在很多重复的公共代码，所有应用共用一套数据库，架构图如下：

将功能A 和功能B 服务化之后，应用作为消费者直接调用服务A 和服务B ，这样就实现了对原有重复代码的收编，同时系统之间的调用关系也更加清晰，如下图：

基于服务注册中心的定于发布机制，实现服务消费者和提供者之间的解耦。

2.2 分而治之

将核心业务抽取出来，作为独立的服务，逐渐形成稳定的底层微服务。
应用的拆分分为水平拆分和垂直拆分两种，水平拆分以业务领域为维度，抽象出几个不同的业务域，每个业务域作为一个独立的服务中心对外提供服务。领域服务可以独立地伸缩和升级，快速地响应需求变化，同时与其它业务领域解耦。原理图如下：

应用的垂直拆分主要包括前后台逻辑拆分、业务逻辑和数据访问层拆分，拆分之后的效果图如下：

2.3 敏捷交付

敏捷性的产生，是将运行中的系统解耦为一系列功能单一服务的结果。微服务架构能够对系统中其它部分的依赖加以限制，这种特性能够让基于微服务架构的应用在应对 BUG 或是对新特性需求时，能够快速地进行变更。而传统的垂直架构：“要对应用程序中某个小部分进行变更，就必须对整体架构进行重新编译和构建，并且重新进行全量部署。”

3 微服务架构解析

微服务架构（MSA）是一种架构风格，旨在通过将功能分解到各个离散的服务中以实现对解决方案的解耦。对比图如下：

3.1 微服务划分原则

通用的划分原则是：微服务通常是简单、原子的微型服务，它的功能单一，只负责处理一件事，与代码行数并没有直接关系，与需要处理的业务复杂度有关。有些复杂的功能，尽管功能单一，但是代码量可能成百上千行，因此不能以代码量作为划分微服务的维度。

“微”所表达的是一种设计思想和指导方针，是需要团队或者组织共同努力找到的一个平衡点。

3.2 开发微服务

对于不同的微服务，虽然实现逻辑不同，但是开发方式、持续集成环境、测试策略和部署机制以及后续的上线运维都是类似的，为了满足 DRY 原则并消除浪费，需要搭建统一的开发打包和持续集成环境。

3.3 基于 Docker 容器部署微服务

Docker 是一套开源工具，它能够以某种方式对现有的基于容器的虚拟化技术进行封装，使得它能够在更广阔的工程社区中得到应用，主要在于快速和可移植性。

3.3.1 快速

普通的虚拟机在每次开机时都需要启动一个完整的新操作系统实例，而 Docker 容器能够通过内核共享的方式，共享一套托管操作系统。这意味着，Docker 容器的启动和停止只需要几百毫秒。这样就有更高的敏捷性。

物理机 VS Docker VS 虚拟机

3.3.2 可移植性

1. 线上线下环境等同性：本地模拟线上环境，定位 BUG 更快。
2. 与特定的云提供商解耦：参考 JVM。
3. 提升运维效率：Docker 对可移植的容器部署进行标准化，节省时间与精力。如果你在构建某个应用程序，你的选择包括物理机、虚拟化的本地基础设施、公有云和私有云，以及各种可用的 PaaS 选项。而通过 Docker 标准化的容器格式，任何一种提供商都可以实现一种统一的部署体验。
4. 敏捷性：快速启动，更敏捷。

3.4 治理和运维微服务

微服务架构对运维和部署流水线要求非常高，服务拆分的粒度越细，运维和治理成本就越高，挑战总结如下：

1. 监控度量问题：海量微服务的各种维度性能 KPI 采集、汇总和分析，实时和历史数据同比和环比，对采集模块的实时性、汇总模块的计算能力、前端运维 Portal 多维度展示能力要求非常高。
2. 分布式运维：服务拆分得越细，一个完整业务流程的调用链就越长，需要采集、汇总和计算的数据量就越大，分布式消息跟踪系统需要能够支撑大规模微服务化后带来的性能挑战。
3. 海量微服务对服务注册中心的处理能力、通知的实时性也带来了巨大挑战。
4. 微服务治理：微服务化之后，微服务相比于传统的 SOA 服务有了指数级增长，服务治理的展示界面、检索速度等需要能够支撑这种变化。
5. 量变引起质变：当需要运维的服务规模达到一定上限后，就由量变引起质变，传统的运维框架架构可能无法支撑，需要重构。

解决微服务运维的主要措施就是：分布式和自动化。利用分布式系统的性能线性增长和弹性扩容能力，支撑大规模微服务对运维系统带来的性能冲击，包括：

1. 分布式性能数据采集、日志采集 Agent。
2. 分布式汇总和计算框架。
3. 分布式文件存储服务。
4. 分布式日志检索服务。
5. 分布式报表展示框架。

3.5 特点总结

1. 单一职责原则：每个服务应该负责单独的功能。
2. 独立部署、升级、扩展和替换。
3. 支持异构/多语言。
4. 轻量级。

因此优点如下：

1. 开发、测试和运维更加简单。
2. 局部修改很容易部署，有利于持续集成和持续交付。
3. 技术选择更灵活，不与特定语言和工具绑定。
4. 有利于小团队作战，敏捷交付。

4 个人总结

微服务涉及到了组织架构、涉及、交付、运维等方面的变革，核心目标是为了解决系统的交付周期，降低维护成本和研发成本。
但是带来了运维成本、服务管理成本等。
不可脱离业务实际而强制使用微服务。

相对于传统的本地 Java API 调用，跨进程的分布式服务调用面临的故障风险更高：

1. 网络类故障：链路闪断、读写超时等。
2. 序列化和反序列化失败。
3. 畸形码流。
4. 服务端流控和拥塞保护导致的服务调用失败。
5. 其它异常。

对于应用而言，分布式服务框架需要具备足够的健壮性，在平台底层能够拦截并向上屏蔽故障，业务只需要配置容错策略，即可实现高可靠性。

1 服务状态监测

在分布式服务调用时，某个服务提供者可能已经宕机，如果采用随机路由策略，消息会继续发送给已经宕机的服务提供者，导致消息发送失败。为了保证路由的正确性，消费者需要能够实时获取服务提供者的状态，当某个服务提供者不可用时，将它从缓存的路由表中删除掉，不再向其发送消息，直到对方恢复正常。

1.1 基于服务注册中心状态监测

以 ZooKeeper 为例，ZooKeeper 服务端利用与 ZooKeeper 客户端之间的长链接会话做心跳检测。

1.2 链路有效性状态监测机制

分布式服务框架的服务消费者和提供者之间默认往往采用长链接，并且通过双向心跳检测保障链路的可靠性。
在一些特殊的场景中，服务提供者和注册中心之间网络可达，服务消费者和注册中心网络也可达，但是服务提供者和消费者之间网络不可达，或者服务提供者和消费者之间链路已经断连。此时，服务注册中心并不能检测到服务提供者异常，但是如果消费者仍旧向链路中断的提供者发送消息，写操作将会失败。

为了解决该问题，通常需要使用服务注册中心检测 + 服务提供者和消费者之间的链路有效性检测双重检测来保障系统的可靠性，它工作原理如下：

当消费者通过双向心跳检测发现链路故障之后，会主动释放链接，并将对应的服务提供者从路由缓存表中删除。当链路恢复之后，重新将恢复的故障服务提供者地址信息加入地址缓存表中。

2 服务健康度监测

在集群组网环境下，由于硬件性能差异、各服务提供者的负载不均等原因，如果采用随机路由分发策略，会导致负载较重的服务提供者不堪重负被压垮。
利用服务的健康度监测，可以对集群的所有服务实例进行体检，根据体检加过对健康度做打分，得分较低的亚健康服务节点，路由权重会被自动调低，发送到对应节点的消息会少很多。这样实现“能者多劳、按需分配”，实现更合理的资源分配和路由调度。

服务的健康度监测通常需要采集如下性能 KPI 指标：

1. 服务调用时延。
2. 服务 QPS。
3. 服务调用成功率。
4. 基础资源使用情况，例如堆内存、CPU 使用率等。

原理如下：

3 服务故障隔离

分为四个层次：

1. 进程级故障隔离
2. VM 级故障隔离
3. 物理机故障隔离
4. 机房故障隔离

3.1 进程级故障隔离

个人理解为线程级。即通过将服务部署到不同的线程池实现故障隔离。对于订单、购物车等核心服务可以独立部署到一个线程池中，与其它服务做线程调度隔离。对于非核心服务，可以合设共享同一个/多个线程池，防止因为服务数过多导致线程数过度膨胀。

服务发布的时候，可以指定服务发布到哪个线程池中，分布式服务框架拦截 Spring 容器的启动，解析 XML 标签，生成服务和线程池的映射关系，通信框架将解码后的消息投递到后端时，根据服务名选择对应的线程池，将消息投递到映射线程池的消息队列中。
原理图如下：

如果故障服务发生了内存泄漏异常，它会导致整个进程不可用。

3.2 VM 级故障隔离

将基础设施层虚拟化、服务化，将应用部署到不同的 VM 中，利用 VM 对资源层的隔离，实现高层次的服务故障隔离，工作原理如下：

3.3 物理机故障隔离

当组网规模足够大、硬件足够多的时候，硬件的故障就由小概率事件转变为普通事件。如何保证在物理机故障时，应用能够正常工作，是一个不小的挑战。
利用分布式服务框架的集群容错功能，可以实现位置无关的自动容错，工作原理如下：

如果要保证当前服务器宕机时不影响部署在上面运行的服务，需要采用分布式集群部署，而且要采用非亲和性安装：即服务实例需要部署到不同的物理机上，通常至少需要 3 台物理机，假如单台物理机的故障发生概率为 0.1 %，则 3 台同时发生故障的概率为 0.001%，服务的可靠性将会达到 99.999%，完全可以满足大多数应用场景的可靠性要求。
物理机故障重启之后，通过扩展插件通知 Watch Dog 重新将应用拉起，应用启动时会重新发布服务，服务发布成功之后，故障服务器节点就能重新恢复正常工作。 ·`

3.4 机房故障隔离

同城容灾时，都需要使用多个机房，下面针对跨机房的容灾和故障隔离方案进行探讨。

机房1 和机房2 对等部署了2套应用集群，每个机房部署一套服务注册中心集群，服务订阅和发布同时针对两个注册中心，对于机房1 或者机房2 的 Web 应用，可以同时看到两个机房的服务提供者列表。
理由时，优先访问同一个机房的服务提供者，当本机房的服务提供者大面积不可用或者全部不可用时，根据跨机房路由策略，访问另一个机房的服务提供者，待本机房服务提供者集群恢复到正常状态之后，重新切换到本机房访问模式。
当整个机房宕机之后，由前端的 SLB\F5 负载均衡器自动将流量切换到容灾机房，由于主机房整个瘫掉了，容灾机房的消费者通过服务状态监测将主机房的所有服务提供者从路由缓存表中删除，服务调用会自动切换到本机房调用模式，实现故障的自动容灾切换。

上面的方案需要分布式服务框架支持多注册中心，同一个服务实例，可以同时注册到多个服务注册中心中，实现跨机房的服务调用。两个机房共用一套服务注册中心也可以，但是如果服务注册中心所在的机房整个宕掉，则分布式服务框架的服务注册中心将不可用。已有的服务调用不受影响，新的依赖服务注册中心的操作江辉失败，例如服务治理、运行期参数调整、服务的状态监测等功能将不可用。

4 其它可靠性特性

4.1 服务注册中心

服务注册中心需要采用对等集群设计，任意一台宕机之后，需要能够自动切换到下一台可用的注册中心。例如 ZooKeeper ，如果某个 Leader 节点宕机，通过选举算法会重新选举出一个新的 Leader，只要集群组网实例数不小于 3，整个集群就能够正常工作。

4.2 监控中心

监控中心集群宕机之后，只丢失部分采样数据，依赖性能 KPI 采样数据的服务健康度监测功能不能正常使用，服务提供者和消费者依然能够正常运行，业务不会中断。

4.3 服务提供者

某个服务提供者宕机之后，利用集群容错策略，会舱室不同的容错恢复手段，例如使用 FailOver 容错策略，自动切换到下一个可用的服务，直到找到可用的服务为止。
如果整个服务提供者集群都宕机，可以利用服务放通、故障引流、容灾切换等手段。

5 个人总结

任何假设的宕机情况都会出现，解决手段不外乎：

1. 对等集群（例如跨机房）。
2. 服务放通（远程错误直接切换为本地调用）。
3. 隔离。

0 前言

关于 ZooKeeper实现分布式锁，笔者在武汉小米一面（结果挂了）被问到过，因此记录如下。

以下的理论知识源自《 从Paxos到Zookeeper分布式一致性原理与实践 》第六章，代码 完全根据书本理论进行实现，并且经多线程测试，在正常情况可行。

源码：https://github.com/LiWenGu/MySourceCode/tree/master/example/src/main/java/com/lwg/zk_project

1 ZooKeeper实现排他锁

1.1 原理

核心点：

1. 抢占式创建相同名称的临时节点，谁成功创建节点，则代表谁获得了锁。
2. 没有创建成功该节点，并且该节点存在，则对该名称的节点进行删除监听。
3. 如果该节点被删除了，则继续重复第 1步。

1.2 流程图

原书流程图：

我自己理解的流程：

1.3 代码实现

统一接口：

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

/**
 * @Author liwenguang
 * @Date 2018/6/15 下午9:16
 * @Description
 */
public interface DistributedLock {

    /**
     * @Author liwenguang
     * @Date 2018/6/15 下午9:17
     * @Description 获取锁，默认等待时间
     */
    default void tryRead() throws ZkException { throw new RuntimeException("子类不支持"); }

    /**
     * @Author liwenguang
     * @Date 2018/6/15 下午9:18
     * @Description 获取锁，指定超时时间
     */
    default void tryRead(long time, TimeUnit unit) { throw new RuntimeException("子类不支持"); }

    void tryWrite() throws ZkException;

    default void tryWrite(long time, TimeUnit unit) { throw new RuntimeException("子类不支持"); }

    /**
     * @Author liwenguang
     * @Date 2018/6/15 下午9:18
     * @Description 释放锁
     */
    void release() throws ZkException;

}

核心代码：

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

private void tryGetLock() {
    CountDownLatch countDownLatch = new CountDownLatch(1);
    while (true) {
        try {
            zkClient.createEphemeral(EXCLUSIVE_LOCK_NAMESPACE + lockPath);
            log.info(Thread.currentThread().getName() + "获取锁成功");
            break;
        } catch (ZkNodeExistsException e) {
            // log.warn(Thread.currentThread().getName() + "获取锁失败");
            if (zkClient.exists(EXCLUSIVE_LOCK_NAMESPACE + lockPath)) {
                MyIZkDataListener myIZkChildListener = new MyIZkDataListener(countDownLatch);
                zkClient.subscribeDataChanges(EXCLUSIVE_LOCK_NAMESPACE + lockPath, myIZkChildListener);
            } else {
                countDownLatch.countDown();
            }
        }
        try {
            // 这里需要阻塞式通知，因此使用 countDownLatch实现
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    log.info("获取到了锁");
}

class MyIZkDataListener implements IZkDataListener {

    private CountDownLatch countDownLatch;


    public MyIZkDataListener(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void handleDataChange(String dataPath, Object data) throws Exception { }

    @Override
    public void handleDataDeleted(String dataPath) throws Exception {
        //log.info(Thread.currentThread().getName() + "被回调了");
        zkClient.unsubscribeDataChanges(EXCLUSIVE_LOCK_NAMESPACE + lockPath, this);
        countDownLatch.countDown();
    }
}

2 ZooKeeper共享锁

2.1 原理

核心点：

1. 无论是读请求（读锁）还是写请求（写锁）都进行创建顺序临时节点，只看后缀的数字我们可以理解为 一种从小到大的队列（例：我们在做订单请求的时候，对订单A做创建-> 支付-> 完成三个操作，对应 ZK节点则节点A下有三个子节点，这时候节点A可以理解为一个队列）。
   
2. 创建完成之后，对读锁，则判断该队列之前是否有写锁，如果有写锁，则对写锁做删除监听。对写锁，判断队列之前是否有锁，如果有锁，则对序号最大的锁做删除监听。
3. 删除监听触发，获取该锁节点下所有的子节点（一个节点即代表锁），重复第 2步。

2.2 流程图

原书流程图：

我自己理解的流程：

2.3 代码实现

核心代码：

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

@Override
public void tryRead() throws ZkException {
    if (!zkClient.exists(SHARED_LOCK_NAMESPACE + lockPath)) {
        zkClient.createPersistent(SHARED_LOCK_NAMESPACE + lockPath);
    }
    CountDownLatch countDownLatch = new CountDownLatch(1);
    curNode = zkClient.createEphemeralSequential(SHARED_LOCK_NAMESPACE + lockPath + "/" + SHARED_READ_PRE, null);
    String curSequence = curNode.split(SHARED_READ_PRE)[1];
    log.info(curSequence + "创建读锁-R");
    while (true) {
        List<String> children = zkClient.getChildren(SHARED_LOCK_NAMESPACE + lockPath);
        // 记录序号比自己小的写请求
        List<String> writers = new ArrayList<>();
        for (String brother : children) {
            if (brother.startsWith(SHARED_WRITE_PRE)) {
                String sequence = brother.split(SHARED_WRITE_PRE)[1];
                if (curSequence.compareTo(sequence) > 0) {
                    writers.add(brother);
                }
            }
        }
        if (writers.isEmpty()) {
            // 没有比自己序号小的写请求，说明自己获取到了读锁
            //log.info(Thread.currentThread().getName() + "没有比自己序号小的写请求-R");
            break;
        } else {
            // 获取最近的那个写锁
            String lastWriter = SHARED_LOCK_NAMESPACE + lockPath + "/" + writers.get(writers.size() - 1);
            // 判断最近的那个写锁期间是否已经释放了
            if (zkClient.exists(lastWriter)) {
                MyReadIZkChildListener myReadIZkChildListener = new MyReadIZkChildListener(lastWriter, countDownLatch);
                zkClient.subscribeDataChanges(lastWriter, myReadIZkChildListener);
            } else {
                countDownLatch.countDown();
            }
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    log.info("获取到了锁-R");
}

class MyReadIZkChildListener implements IZkDataListener {

    private String lastWriter;
    private CountDownLatch countDownLatch;

    public MyReadIZkChildListener(String lastWriter, CountDownLatch countDownLatch) {
        this.lastWriter = lastWriter;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void handleDataChange(String dataPath, Object data) throws Exception { }

    @Override
    public void handleDataDeleted(String dataPath) throws Exception {
        //log.info(Thread.currentThread().getName() + "比自己序号小的那个写请求被释放了-R");
        zkClient.unsubscribeDataChanges(lastWriter, this);
        // 最近的那个写锁被释放了，但是不排除释放过程中，有其它写锁新加入，因此读锁需要重新获取列表
        countDownLatch.countDown();
    }
}

@Override
public void tryWrite() throws ZkException {
    CountDownLatch countDownLatch = new CountDownLatch(1);
    if (!zkClient.exists(SHARED_LOCK_NAMESPACE + lockPath)) {
        zkClient.createPersistent(SHARED_LOCK_NAMESPACE + lockPath);
    }
    curNode = zkClient.createEphemeralSequential(SHARED_LOCK_NAMESPACE + lockPath + "/" + SHARED_WRITE_PRE, null);
    String curSequence = curNode.split(SHARED_WRITE_PRE)[1];
    log.info(curSequence + "创建写锁-W");
    while (true) {
        List<String> children = zkClient.getChildren(SHARED_LOCK_NAMESPACE + lockPath);
        // 记录序号比自己小的请求
        List<String> writersOrReader = new ArrayList<>();
        for (String brother : children) {
            if (brother.equals(SHARED_WRITE_PRE + curSequence)) {
                // 排除自己
                continue;
            }
            String sequence = "";
            if (brother.contains(SHARED_WRITE_PRE)) {
                sequence = brother.split(SHARED_WRITE_PRE)[1];
            } else if (brother.contains(SHARED_READ_PRE)) {
                sequence = brother.split(SHARED_READ_PRE)[1];
            } else {
                // 异常名称节点的处理
            }
            if (curSequence.compareTo(sequence) > 0) {
                writersOrReader.add(brother);
            }
        }
        if (writersOrReader.isEmpty()) {
            // 没有比自己序号小的请求，说明自己获取到了读锁
            //log.info(Thread.currentThread().getName() + "没有比自己序号小的请求-W");
            break;
        } else {
            // 获取最近的那个锁
            String lastWriterOrReader = SHARED_LOCK_NAMESPACE + lockPath + "/" + writersOrReader.get(writersOrReader.size() - 1);
            // 判断最近的那个锁期间是否已经释放了
            if (zkClient.exists(lastWriterOrReader)) {
                MyWriteIZkChildListener myWriteIZkChildListener = new MyWriteIZkChildListener(lastWriterOrReader, countDownLatch);
                zkClient.subscribeDataChanges(lastWriterOrReader, myWriteIZkChildListener);
            } else {
                countDownLatch.countDown();
            }
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    log.info("获取到了锁-W");
}

class MyWriteIZkChildListener implements IZkDataListener {

    private String lastWriterOrReader;
    private CountDownLatch countDownLatch;

    public MyWriteIZkChildListener(String lastWriterOrReader, CountDownLatch countDownLatch) {
        this.lastWriterOrReader = lastWriterOrReader;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void handleDataChange(String dataPath, Object data) throws Exception {
    }

    @Override
    public void handleDataDeleted(String dataPath) throws Exception {
        //log.info(Thread.currentThread().getName() + "比自己序号小的那个请求被释放了，循环-W");
        zkClient.unsubscribeDataChanges(lastWriterOrReader, this);
        countDownLatch.countDown();
    }
}

3. 待续

readlock

随着业务分布式架构的发展，系统间的系统调用日趋复杂，以电商的商品购买为例，前台界面的购买操作设计到底层上百次服务调用，涉及到的中间件包括：

1. 分布式服务框架
2. 消息队列
3. 分布式缓存
4. 分布式数据访问中间件
5. 分布式文件存储系统
6. 分布式日志采集
7. 其它……

如果无法有效清理后端的分布式调用和依赖关系，故障定界将会非常困难。利用分布式消息跟踪系统可以有效解决服务化之后系统面临的运维挑战，提高运维效率。

1 业务场景分析

以下为分布式调用示意图：

1.1 故障的快速定界定位

传统应用软件发生故障时，往往通过接口日志手工从故障节点采集日志进行问题分析定位，分布式服务化之后，一次业务调用可能涉及到后台上百次服务调用，每个服务又是集群组网，传统人工到各个服务节点人肉搜索的方式效率很低。
希望能够通过调用链跟踪，将一次业务调用的完整轨迹以调用链的形式展示出来，通过图形化界面查看每次服务调用结果，以及故障信息。

通过在业务日志中增加调用链 ID ，可以实现业务日志和调用链的动态关联。通过调用链进行快速故障定界，然后通过 ID 关联查询，可以快速定位到业务日志相关信息。

1.2 调用路径分析

通过对调用链调用路径的分析，可以识别应用的关键路径：应用被调用得最多的入口、服务是哪些，找出服务的热点、耗时瓶颈和易故障点。同时为性能优化、容量规划等提供数据支撑。

1.3 调用来源和去向分析

通过调用去向分析，可以对服务的依赖关系进行梳理：

1. 应用直接和间接依赖了哪些服务。
2. 各层次依赖的调用时延、QPS、成功率等性能 KPI指标。
3. 识别不合理的强依赖，或者冗余依赖，反向要求开发进行依赖解耦和优化。

通过对调用来源进行 TOP排序，识别当前服务的消费来源，以及获取各消费者的 QPS、平均时延、出错率等，针对特定的消费者，可以做针对性治理，例如针对某个消费者的限流降级、路由策略修改等，保障服务的 SLA。

2 分布式消息跟踪系统设计

消息跟踪系统的核心就是调用链：每次业务请求都生成一个全局唯一的 TraceID,通过跟踪 ID将不同节点间的日志串接起来，形成一个完整的日志调用链，通过对调用链日志做实时采集、汇总和大数据分析，提取各种维度的价值数据，为系统运维和运营提供大数据支撑。

2.1 系统架构

分布式消息跟踪系统的整体架构如下，由四部分组成：

1. 调用链埋点日志生成
2. 分布式采集和存储埋点日志
3. 在线、离线大数据计算，对调用链数据进行分析和汇总
4. 调用链的界面展示、排序和检索等

2.2 埋点日志

埋点就是分布式消息跟踪系统在当前节点的上下文信息，埋点可以分为两类：

1. 客户端埋点，客户端发送请求消息时生成的调用上下文，通常包括 TraceID、调用方 IP、调用方接口或者业务名称、调用的发起时间、被调用的服务名、方法名、IP 地址和端口等信息。
2. 服务端埋点，服务端返回应答消息时在当前节点生成的上下文，包括 TraceID、调用方上下文信息、服务端处理的耗时、处理结果等信息。

埋点日志的实现，通常会包含如下几个功能：

1. 埋点规范，主要用于业务二次定制开发和第三方中间件/系统对接。
2. 埋点日志类库，服务生成埋点上下文，打印埋点日志等。
3. 中间件预置埋点功能，应用不需要开发任何业务代码即可直接使用，也可以通过埋点类库将应用自身的业务字段携带到调用链上下文中，例如终端类型、手机号等。

消息跟踪 ID 通常由调用首节点负责生成（各种门户 Portal），本 JVM 之内通常线程上下文传递 TraceID，跨节点传递时，往往通过分布式服务框架的显式传参传递到下游节点，实现消息跟踪上下文的跨节点传递。
埋点日志上下文通常需要包含如下内容：

1. TraceID、RPCID、调用的开始时间、调用类型、协议类型、调用方 IP 和端口、被调用方 IP 和端口、请求方接口名、被调用方服务名等信息。
2. 调用耗时、调用结果、异常信息、处理的消息报文大小等。
3. 可扩展字段，通常用于应用扩展埋点上下文信息。

消息跟踪ID（TraceID）是关联一次完整应用调用的唯一标识，需要在整个集群内唯一，它的取值策略有很多，例如 UUID，UUID（Universally Unique Identifier）即全局唯一标识符，是指在一台机器上生成的数字，它保证对在同一时空中所有机器都是唯一的。按照开发软件基金会（OSF）制定的标准计算，用到了以太网卡地址、纳秒级时间、芯片 ID 码和许多可能的数字。由以下几部分组合：当前日期和时间（UUID 的第一部分与时间有关，如果你在生成一个 UUID 之后，过几秒又生成一个 UUID，则第一部分不同，其余相同），时钟序列，全局唯一的 IEEE 机器识别号（如果有网卡，从网卡获得，没有网卡以其它方式获得），UUID 的唯一缺陷在于生成的结果串会比较长。

1. IP 地址和端口：调用发起方和被调用方 IP 地址、端口号
2. 时间戳：埋点上下文的生成时间
3. 顺序号：标识链路传递序列的 RpcID
4. 进程号：应用的进程 ID
5. 随机数：例如可以选择 8 位数的随机数

原理上，埋点日志比较简单，实现起来并不复杂。但是在实际工作中，埋点日志也会面临一些技术挑战，举例如下：

1. 异步调用：业务服务中直接调用 MQ 客户端，或者其它中间件的客户端时，可能会发生线程切换，通常线程上下文传递的埋点信息丢失，MQ 客户端会认为自己是首节点，重新生成 TraceID，导致调用链串接不起来。
2. 性能影响：由于 Java I/O 操作通常都是同步的，如果磁盘的 WIO 比较高，会导致写埋点日志阻塞应用线程，导致时延增大。频繁地写埋点日志，也会占用大量的 CPU、带宽等系统资源，影响正常业务的运行。

对于线程切换问题，在切换时需求做线程上下文的备份，将埋点上下文复制到切换的线程上下文中，即可解决问题。
频繁写埋点日志影响性能问题，可以通过如下措施改善该问题：

1. 支持异步写日志，防止写埋点日志慢阻塞服务线程。具体实现上可以通过采用 log4j 的异步 Appender、独立的日志线程池甚至是 JDK1.7 之后提供的异步文件操作接口。
2. 提供可灵活配置的埋点采样率，控制埋点日志量。
3. 批量写日志，日志流控机制。

2.3 采样率

对于高 QPS 的应用，服务调用埋点本身的性能损耗也不容忽视，为了解决 100% 全采样的性能损耗，可以通过采样率来实现埋点低损耗的目标。
采样包括静态采用和动态采样两种，静态采样就是系统上线时设置一个采样率，无论负载高低，均按照采样率执行。动态采样率根据系统的负载可以自动调整，当负载比较低的时候可以实现 100% 全采样，在负载非常重时甚至可以降低到 0 采样。

是否采样由调用链的首节点进行判断，首节点根据采样率算法，决定某个业务访问是否采样，如果需要采样，则把采样标识、TraceID 等采样上下文发送到下游服务节点，下游服务节点根据采样标识做判断，如果采样则获取调用链上下文并补充完整，反之则不埋点。

2.4 采集和存储埋点日志

开源 的 ELK，原理如下：

需要考虑：

1. 采集过程中发生宕机，如何在中断点恢复采集。
2. 采集过程中如果埋点日志发生了文件切换（例如达到单个日志文件 100MB 上限之后，自动进行文件切换），如何正确应对。
3. 采集 Channel 发生网络故障，导致采集的日志部分发送失败，故障恢复之前，日志如何缓存，故障恢复之后，已采集尚未发送的日志如何发送。
4. 考虑到性能，是不是单条采集、批量发送性能更优。

3 个人总结

通过对业务流程的记录和采集，进行在线和离线的大数据计算，数据清洗获取有价值的数据。同时还能根据运行情况做服务的调整。