一本正经的胡说八道

对错自在你心，可能就是个饭后闲扯

所得

以前同事问，什么时候跳槽？或者什么时候会出去看看机会呢？我都会回答做到有所得的时候。有所得，当然不是说拷贝公司几份源码，那可是公司资产，不可误入歧途。但知识是自己的，成长是自己的

一直也是这么践行的，从出校门不懂的书生到能把书本知识灵活运用到一线工作，从毫无经验到开发工业级产品，从职场小白到带队主导开发流程，虽然经历的大多数公司已经倒闭了，但自己是实实在在的成长了。

是的，是所有得再考虑外面机会的，小池塘不能满足成长时，就要去大池塘

前些天跟老同事见面聊游戏开发周期，我问现在立项时还是提议半年开发一款产品吗？同事呵呵两声回答，半年开发一款早就得卷地铺走人了，现在两周就得上线。着实吓了我一跳，原来他们现在是主攻小游戏，依赖完备的微信生态，快速产品，快速变现

去年同事项目出了个爆款，公司整年有近3亿纯利，开发自然奖金不菲，当然也没有想像中那么多

我思考他们的开发模式，直接模仿国外游戏，改头换皮，再稍加修改，上线。大多数产品肯定是上得快，死得也快。如此环境，对技术人员的成长有多少帮助

绝大多数收益进入老板口袋，开发人员一直重复再重复，甚至有些游戏只是改了个名字，最后他们能得到什么？

此类游戏业务简单，公司组织也扁平，尤其服务端，平台提供了各种服务，小游戏前端已经在喊着要干掉服务端的口号，不 管哪个开发岗位，在几年后，当面对新机会时，怎么阐述自己的价值，是抄得更快速？还是因为运气，出了多少爆款？

由他而又思考到自己所谓的大公司，有人评价阿里工程师，说普遍被中间件惯坏了

不是背后说人坏话，只是一种现象，可能是大公司人的普遍问题
，进阿里是大多数技术人的梦想，包括我，但估计此生无缘，能力实在达不到阿里要求

面试造火箭，入职拧螺丝；公司越成熟，基础建设就越完备、基建设施模块化程度越高，工具体验做得越好，平台做得越完善，那么对上层做业务来讲，底层细节也就屏蔽越多

对业务来说本身是件好事，但对业务开发技术人员呢？有些领域技术变化很快，两三年可能迭代了好几轮，在深度、广度上有了质的飞跃，而你该怎么办，削足适履追求新技术吗？

回顾自己技术实力时，脱离了这些基础设施，还有多少生产力？尤其微服务会打散各个业务线，可能一条完整的业务链路都理不清

此时，还能有所得吗？得多少？褪去头顶公司光环，酒桌上的牛还能吹多大？

所思

有人说技术人就像丢在大海里的漂流瓶，努力漂泊，孤傲不羁，却怎么也不能融入大海，装不满自己空空如也的肚腩，因为他们不知道身体倾斜一点，才是最佳姿势，才有最快的装水速度

也许真是远离技术看技术，才有更大的格局

从事游戏开发很多年，一直觉得自己对游戏开发还是有见解的，但好像只是开发阶段的把握，而不是游戏本身的整体把握。

早些年总是抱怨，为什么我们技术这么好，为什么游戏总是死呢？其实考虑面太窄，没想过运营，商务，维护，推广等等事项，手上有把锤子，到处只看到钉子了

对于游戏类型，也很局限，最近看到一个外行人总结游戏类型，特别的汗颜

依据不同的投入类型，我们可以把游戏分成三大类：

1）反应。这类游戏，需要实时观察周围环境，并快速作出操作和应对。最典型的就是各种射击、动作、即时战略类游戏。

2）策略。这类游戏，需要综合考虑各种可能性，对未来作出规划，并安排好下一步行动。典型如各种策略模拟游戏。

3）沉浸。这类游戏，需要代入角色，理解和接受其世界观，阅读、记忆大量剧情，探索故事线。典型就是各种角色扮演游戏。

游戏的玩法可以千奇百怪，类型可以随意组合，但好的游戏，对玩家的要求，基本都不出这三类

这些只是单机玩法，大型网络游戏，还涉及到群体心理等等

在游戏行业打拼多年，尽然没有一个外行总结到位，抽象高度太低太低

技术里的世界不小，但技术外的世界更大

是该放下手中的技术，抬头看看外面的世界

功夫在诗外，也许再回头看技术时，别有一番天

当然千万别一时亢奋放弃技术，远离技术是在追求技术无法再提升格局的时候，跳出来，回头看

是从简入繁完成后，由繁化简的过程中的技法；无法打开一把锁时，不能只盯着锁看，因为钥匙可能在远离锁的地方

那么如何远离技术呢？远离程度呢？也许多读书是个切入口，《关于读书》这篇文章也许对你有所帮助，如果能找到牛人指点更好。

通过《zookeeper知识结构1》了解了zookeeper是什么？为什么使用zookeeper? 以及zookeeper内部数据结构，选举机制

zab定义

ZAB全称ZooKeeper Atomic Broadcast protocol

ZooKeeper原子广播协议，实现了主备模式下的系统架构，保持集群中各个副本之间的数据同步(数据最新，一致性)

原子广播

在具体深入zab之前，先搞明白原子广播

原子性好理解，不管是事务的ACID，还是多线程中，都有这个概念；广播也好理解，像系统中常引入消息队列进行消息广播

但两者合在一起，有点犯晕了，广播怎么需要原子性？哪部分操作不可分割呢？

所有事务请求的处理结果在整个集群中所有机器上的应用情况是一致的,也就是说,要么整个集群所有机器都成功应用了某一个事务, 要么就都没应用.

按照上面的解释，原子性体现在所有机器事务一致性,要么都接受广播，要么都不接受；为了更深入理解，从头开始深入一下这个机制的大致流程

在分布式中，这个机制有很多具体实现，比如2PC,3PC，paxos等等

副本

在这个机制介绍前，需要先介绍下分布式中的副本(Replication)

副本简直是处理故障恢复的的万能钥匙

数据副本的收益：

1. 提升系统可用性，需要挂更多的节点才会导致数据丢失
2. 提升系统性能，多个副本可以同时处理或者交给更快的机器处理

分布式系统采用副本可以获得可扩展性，高性能，可用性，容错性

1. 害怕数据不可用，采用副本吧，多副本能确保数据由于故障丢失的概率大大降低；
2. 计算太慢，采用副本吧，将计算散布到多台机器上；
3. I/O太慢，采用副本吧，将数据cache到local机器上，可以极大的提升吞吐

正常多副本，可以从任何副本请求数据

数据同步

为了多个副本上的数据同步，一般会加一个dispatche,client请求到dispatche上，dispatche会对请求进行排序，并按同样的顺序分发到各个副本，保持各副本数据一致

此时，dispatche成了系统中的单点，不具备高可用，所以会部署多个dispatch

每个dispatche相互通信，并且每个dispatche都可以分发到各个副本；此时就会出现上图中的并发更新一致性问题，类似于《深入浅出事务》中的第二类更新丢失

在并发操作时，应用本身对于数据的先后都是可以接受的，但在各个副本中的数据必须保持一致，也就是下图中的结果都是可以的

也就是讲，各个dispatche得以相同的顺序进行更新副本

顺序同步

解决并发更新不一致问题，思路就是对请求操作进行排序，按顺序执行

数据更新的过程如下：

1. 客户发送操作请求到任意一个节点的分发器上
2. 分发器接收到请求后，将请求广播到其他节点上的分发器，并且这些分发器之间会对所有的并发请求进行排序。最终每个节点的分发器上都会有一份完全一样的请求列表。这个功能通常称作原子广播（Atomic Broadcast） 或者 全局排序广播（Total Order Broadcast)
3. 分发器将列表中的操作请求按照顺序送给本节点的数据副本

在这个模型中，原子广播的逻辑和业务逻辑是分开的。这么做的好处是非常明显的，业务逻辑的实现不再受分布式需求的限制，原子广播的逻辑则不需要考虑业务逻辑的具体需求。独立的原子广播的逻辑可以被重用到很多的分布式的应用上

原型

根据上面的推导，得出一个最简单的原型

dispatche增加一个队列，也可以把它设计成日志文件（顺序追加的文件）或者管道等等，当有请求时，会按顺序存储到队列，但这个队列中的每个位置只能存储一次数据，存储后不能进行更改

请求操作

1. 收到客户端的数据存储请求后，选择一个存储位置。发送数据储存指令给其他的分发器，同时将数据存储到自己的存储队列中
2. 当收到其他分发器发送的存储指令后，将数据存储到自己的存储队列。如果该位置已经存储了数据，则返回失败

• 数据存储指令的内容:[存储位置，数据]
• 存储位置的选择:选择最小的空存储位

并发更新

当两个client分别发到不同到dispatche上请求，两个dispatche相互同步时，出现了不一致，违背了各个队列的每个位置只能存储一次元素，不能更改的原则

锁

对于并发问题，首先想到就是加锁，对于分布式系统怎么加锁呢？

2PC提供了好的思路

借鉴一下2pc，加锁过程如下：

第一步：加锁

1. 接受到请求，选择本地队列位置，并加锁
2. 发送加锁指令到别的dispatche2，dipatche2进行相同位置加锁，并回返lock ack

• 加锁指令：[位置]

第二步：更新

1. 收到所有dispatche都返回lock ack
2. 数据写入队列位置上，并发送存储指令给所有disptache
3. dispatche收到存储指令，把数据写入队列相应位置

死锁

并发考虑的两个问题：安全性与活跃性

dispatche1与dispatche2分别接收到请求，在第一步加锁时，先给自己队列加锁，再发送加锁指令给其它dispatche

此时，就出现死锁：如dispathe1对自身队列位置1加锁，再发送指令给disptache2的队列位置1加锁，但disptache2的队列位置1被自己锁住了，反之disptache1队列位置1也一样，此时双方进入死锁，永远阻塞

优先级锁

有一种策略，就是给锁增加一个优化级

• 每个锁请求都有一个优先级，优先级高的锁请求可以撤销优先级低的锁。
• 如果一个存储指令的锁被撤销了，就不能被执行

如图所示，两个dispatche分别接收到请求，p1 p2是各自锁的优先级，p1高于p2

1. dispatch1接收请求后，自己队列加锁；收到dispatch2的加锁请求，但p2优先级低于p1，所以被阻塞
2. dispatche2自身加锁后，收到dispatche1的加锁请求，由于p1高于p2，dispatche2撤销p2操作，p1成功加锁
3. dispatche1成功在所有dispatche上加锁

优先级：

1. 有先后次序,优先级不能相等
2. 不可重复

比如以dispatche的id作为优先级值，dispatcheId是不同的，也能区分优先级，但这样会出现不平衡，以id大的优先级高，那永远id为大的优先获取锁，加锁成功处理请求

为了更好的均衡各个客户的请求处理，可以采用下面的优先级定义：[数值，dispatcheID]

数值可以由各个dispatche指定，比如所有disptache都原子累加，这样可以保证每个dispathe的均衡，当数值相等时，再比较dispatcheId

节点故障

故障可能发生在每个阶段，一是加锁阶段，二是数据更新阶段；再细节大概有三种情况：

1. 加锁成功，但没有写入数据
2. 部分节点写入了数据
3. 故障dispatche已经写入数据，但没有同步到别的dispatche

加锁成功，没有写入数据

这个很好解决，通过优先级锁，另的节点发起一个更高优先级操作就可以覆盖先前的记录

部分节点写入数据

解决这个问题，需要加强一下更新操作，之前的更新操作，发送存储指令给各个节点，就结束了

现在不单单发送指令，还需要再广播

也就是节点接受到存储指令，如果节点已经有数据写入，则与数据一起返回；这样当所有节点返回加锁成功后，检查是否有数据返回，如果有数据返回，则将数据放入存储指令，发送给所有节点

没有同步别的dispatche

需要增加一个“预存储队列”，预写入机制

当dispatche1发生故障时，其它的dispatche的预存储队列中已经存入了数据。其它节点接管dispatche1时，会先重发预存储队列中的数据到所有dispatche

预写入过程可以保证：如果数据被写入了任意的存储队列，那么所有节点的预存储队列都有这个数据

多数派

上面的广播机制中，加锁以及预写入都需要所有节点返回成功。如果任意一个节点有故障都会失败。在复杂网络环境下，整个系统很脆弱，不能高可用

因此可以改进为半数以上节点成功回复就可以

大多数派机制下，会带来一些更复杂的中间状态，整个过程：

1. 发送加锁指令
2. 收到加锁指令后，检查指定存储位置是否已经加锁，如没有，则返回加锁成功；如被加锁，则比较锁优先级，如果优先级更高，就撤销原有锁，重新加锁返回成功；如果已经预写入数据，则将数据一并返回
3. 当超过半数节点返回加锁成功后，检查是否有数据返回。如果有数据返回，则将优先级最高的数据存入预存储指令。如果没有数据返回，则将自己的数据写入预存储指令。发送预存储指令给所有dispatche,并写入自己的预存储队列
4. 收到预存储指令，将数据写入预存储队列。如果预存储锁被撤销，则返回失败
5. 当超过半数返回预存储成功，刚发送存储指令给所有dispatche，并写入自己的存储队列
6. 当收到存储队列，将数据写入自己存储队列中

ZAB详细

有了上面的原型，理解其它的具体协议就会轻松很多，在具体实现时，都会看到原型中的概念

ZAB协议是为分布式协调服务 Zookeeper 专门设计的一种支持 崩溃恢复 和 原子广播 协议

ZAB协议定义了选举（election）、发现（discovery）、同步（sync）、广播(Broadcast)四个阶段

原型抽象

根据上面的原型，结合zk的源码，梳理一下源码中对应原型的抽象

投票对象

这个对象对应着原型中的存储指令，优先级加锁指令

1
2
3
4
5
6
7

public Vote(long id, long zxid) {
    this.id = id;
    this.zxid = zxid;
    this.electionEpoch = -1;
    this.peerEpoch = -1;
    this.state = ServerState.LOOKING;
}

id：被推举的Leader的SID

zxid：被推举的Leader事务ID

为了保证事务的顺序一致性，zookeeper 采用了递增的事 务 id 号（zxid）来标识事务。

所有的提议（proposal）都在被提出的时候加上了 zxid。

实现中 zxid 是一个 64 位的数字

它高32 位是 epoch（ZAB 协议通过 epoch 编号来区分 Leader 周期变化的策略）用来标识 leader关系是否改变，每次一个 leader 被选出来，它都会有一个新的epoch=（原来的 epoch+1），标识当前属于那个 leader 的统治时期。

低 32 位用于递增计数

可以想象为中国古代的年号，例如万历十五年，万历是epoch，十五年是id

electionEpoch：逻辑时钟，用来判断多个投票是否在同一轮选举周期中，该值在服务端是一个自增序列，每次进入新一轮的投票后，都会对该值进行加1操作

peerEpoch：被推举的Leader的epoch

electionEpoch和peerEpoch的区别在于，electionEpoch记录的选举的轮次，而peerEpoch则指的是当前leader的任期

state：当前服务器的状态

FastLeaderElection

zk默认的选举算法，为什么需要选举可以参照《zookeeper知识结构1》

术语

• 外部投票：特指其他服务器发来的投票。
• 内部投票：服务器自身当前的投票。
• 选举轮次：Zookeeper服务器Leader选举的轮次，即logicalclock
• PK：对内部投票和外部投票进行对比来确定是否需要变更内部投票

electionEpoch和logicalclock的区别在于，electionEpoch指的是发出server的logicalclock，而logicalclock则指的是当前Server所处的选举的轮次

队列

• sendqueue：选票发送队列，用于保存待发送的选票。
• recvqueue：选票接收队列，用于保存接收到的外部投票。
• WorkerReceiver：选票接收器。其会不断地从QuorumCnxManager中获取其他服务器发来的选举消息，并将其转换成一个选票，然后保存到recvqueue中，在选票接收过程中，如果发现该外部选票的选举轮次小于当前服务器的，那么忽略该外部投票，同时立即发送自己的内部投票。
• WorkerSender：选票发送器，不断地从sendqueue中获取待发送的选票，并将其传递到底层QuorumCnxManager中

QuorumCnxManager

ClientCnxn是ZooKeeper客户端中用于处理网络I/O的一个管理器

在Leader选举的过程中也有类似的角色，那就是QuorumCnxManager——每台服务器启动的时候都会启动一个QuorumCnxManager，负责各台服务器之间的底层Leader选举过程中的网络通信

QuorumCnxManager这个类内部维护了一系列的队列，用于保存接收到的、待发送的消息，以及消息的发送器。

除接收队列以外，这里提到的所有队列都有一个共同点——按SID分组形成队列集合，我们以发送队列为例来说明这个分组的概念。

假设集群中除自身外还有4台机器，那么当前服务器就会为这4台服务器分别创建一个发送队列，互不干扰。

• queueSendMap：消息发送队列，用于保存那些待发送的消息。queueSendMap是一个Map，按照SID进行分组，分别为集群中的每台机器分配了一个单独队列，从而保证各台机器之间的消息发送互不影响。
• senderWorkerMap：发送器集合。每个SendWorker消息发送器，都对应一台远程ZooKeeper服务器，负责消息的发送。同样，在sendWorkerMap中，也按照SID进行了分组。
• lastMessageSent：最近发送过的消息。在这个集合中，为每个SID保留最近发送过的一个消息

在SendWorker的具体实现中，有一个细节需要我们注意一下：一旦ZooKeeper发现针对当前远程服务器的消息发送队列为空，那么这个时候就需要从lastMessageSent中取出一个最近发送过的消息来进行再次发送。这个细节的处理主要为了解决这样一类分布式问题：接收方在消息接收前，或者是在接收到消息后服务器挂掉了，导致消息尚未被正确处理。那么如此重复发送是否会导致其他问题呢？当然，这里可以放心的一点是，ZooKeeper能够保证接收方在处理消息的时候，会对重复消息进行正确的处理

lastMessageSent接近原型中的预存储队列

选票过程

• 1.自增选举轮次。Zookeeper规定所有有效的投票都必须在同一轮次中，在开始新一轮投票时，会首先对logicalclock进行自增操作。
• 2.初始化选票。在开始进行新一轮投票之前，每个服务器都会初始化自身的选票，并且在初始化阶段，每台服务器都会将自己推举为Leader
• 3.发送初始化选票。完成选票的初始化后，服务器就会发起第一次投票。Zookeeper会将刚刚初始化好的选票放入sendqueue中，由发送器WorkerSender负责发送出去。
• 4.接收外部投票。每台服务器会不断地从recvqueue队列中获取外部选票。如果服务器发现无法获取到任何外部投票，那么就会立即确认自己是否和集群中其他服务器保持着有效的连接，如果没有连接，则马上建立连接，如果已经建立了连接，则再次发送自己当前的内部投票
• 5.判断选举轮次。在发送完初始化选票之后，接着开始处理外部投票。在处理外部投票时，会根据选举轮次来进行不同的处理。
  • 5.1.外部投票的选举轮次大于内部投票。若服务器自身的选举轮次落后于该外部投票对应服务器的选举轮次，那么就会立即更新自己的选举轮次(logicalclock)，并且清空所有已经收到的投票，然后使用初始化的投票来进行PK以确定是否变更内部投票。最终再将内部投票发送出去。
  • 5.2.外部投票的选举轮次小于内部投票。若服务器接收的外选票的选举轮次落后于自身的选举轮次，那么Zookeeper就会直接忽略该外部投票，不做任何处理，并返回步骤4。
  • 5.3.外部投票的选举轮次等于内部投票。此时可以开始进行选票PK
• 6.选票PK。在进行选票PK时，符合任意一个条件就需要变更投票
  • 6.1.若外部投票中推举的Leader服务器的选举轮次大于内部投票，那么需要变更投票。
  • 6.2.若选举轮次一致，那么就对比两者的ZXID，若外部投票的ZXID大，那么需要变更投票。
  • 6.3.若两者的ZXID一致，那么就对比两者的SID，若外部投票的SID大，那么就需要变更投票。
• 7.变更投票。经过PK后，若确定了外部投票优于内部投票，那么就变更投票，即使用外部投票的选票信息来覆盖内部投票，变更完成后，再次将这个变更后的内部投票发送出去。
• 8.选票归档。无论是否变更了投票，都会将刚刚收到的那份外部投票放入选票集合recvset中进行归档。recvset用于记录当前服务器在本轮次的Leader选举中收到的所有外部投票（按照服务队的SID区别，如{(1, vote1), (2, vote2)…}）。
• 9.统计投票。完成选票归档后，就可以开始统计投票，统计投票是为了统计集群中是否已经有过半的服务器认可了当前的内部投票，如果确定已经有过半服务器认可了该投票，则终止投票。否则返回步骤4
• 10.更新服务器状态。若已经确定可以终止投票，那么就开始更新服务器状态，服务器首选判断当前被过半服务器认可的投票所对应的Leader服务器是否是自己，若是自己，则将自己的服务器状态更新为LEADING，若不是，则根据具体情况来确定自己是FOLLOWING或是OBSERVING。

以上10个步骤就是FastLeaderElection的核心，其中步骤4-9会经过几轮循环，直到有Leader选举产生。

总结

通过对原子广播原型的理解，更容易理解zab，对于paxos也一样

当然zab还有很多的细节，还能再深入，挖出很多知识点。但只看理论终归有些空洞，下一篇实践一下，详述zk版本分布式锁

参考资料

Leader选举

分布式系统

由浅入深理解Paxos协议

之前写过关于zookeeper的一篇文章《zookeeper-paxos》，paxos太难理解了，当时理解了，但现在又忘记了，机械学习果然是不行的

虽然曾经有一篇文章讲阿里不使用zk做服务发现，但大多数公司的分布式架构中基本都能看到zk的身影，而且他躲在里面，你可能看不到他，感受不到他的存在

对于架构体系中的这样一位选手，了解，学习，研究是相当有必要的

ZK是什么

ZooKeeper 是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby 一个开源的实现

ZooKeeper 是集群的管理者，监视着集群中各节点的状态，根据节点提交的反馈进行下 一步合理的操作。最终，将简单易用的接口和功能稳定，性能高效的系统提供给用户

zooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services

这大概描述了Zookeeper的作用，配置管理，名字服务，提供分布式同步以及集群管理

为什么需要ZK

知道了zk的定义，其实跟不知道差不多，还是要追根溯源，看看zk今世因缘，存在的意义

zk的历史很多地方都有介绍，这儿就不赘述了

相对历史，更想知道为什么需要zk？

以前经历的系统，都是使用redis做为服务中心了，不管使用single redis,还是redis cluster，能胜任架构需求，全局命名服务、订阅发布监听服务列表、分布式锁足矣

一度怀疑zk的价值，也阻碍了进一步学习的热情，但存在即合理

设计 ZooKeeper 的目的是为了减轻分布式应用程序所承担的协调任务

还是从ZK的定义追溯它的作用

配置管理

在我们的应用中除了代码外，还有一些就是各种配置。比如数据库连接等

一般我们都是使用配置文件的方式，在代码中引入这些配置文件。但是当我们只有一种配置，只有一台服务器，并且不经常修改的时候，使用配置文件是一个很好的做法，但是如果我们配置非常多，有很多服务器都需要这个配置，而且还可能是动态的话使用配置文件就不是个好主意了。

这个时候往往需要寻找一种集中管理配置的方法，我们在这个集中的地方修改了配置，所有对这个配置感兴趣的都可以获得变更。比如我们可以把配置放在数据库里，然后所有需要配置的服务都去这个数据库读取配置

但是，因为很多服务的正常运行都非常依赖这个配置，所以需要这个集中提供配置服务的服务具备很高的可靠性。

一般我们可以用一个集群来提供这个配置服务，但是用集群提升可靠性，那如何保证配置在集群中的一致性呢？ 这个时候就需要使用一种实现了一致性协议的服务了。Zookeeper就是这种服务，它使用Zab这种一致性协议来提供一致性

名字服务

比如为了通过网络访问一个系统，我们得知道对方的IP地址，但是IP地址对人非常不友好，这个时候我们就需要使用域名来访问。

但是计算机是不能是别域名的。

怎么办呢？

如果我们每台机器里都备有一份域名到IP地址的映射，这个倒是能解决一部分问题，但是如果域名对应的IP发生变化了又该怎么办呢？于是我们有了DNS这个东西。

我们只需要访问一个大家熟知的(known)的点，它就会告诉你这个域名对应的IP是什么。

在我们的应用中也会存在很多这类问题，特别是在我们的服务特别多的时候，如果我们在本地保存服务的地址的时候将非常不方便，但是如果我们只需要访问一个大家都熟知的访问点，这里提供统一的入口，那么维护起来将方便得多了

分布式锁

比如在一个分布式环境中，为了提高可靠性，我们的集群的每台服务器上都部署着同样的服务

但是，一件事情如果集群中的每个服务器都进行的话，那相互之间就要协调，编程起来将非常复杂。而如果我们只让一个服务进行操作，那又存在单点。通常还有一种做法就是使用分布式锁，在某个时刻只让一个服务去干活，当这台服务出问题的时候锁释放，立即fail over到另外的服务

这在很多分布式系统中都是这么做，这种设计有一个更好听的名字叫Leader Election(leader选举)。比如HBase的Master就是采用这种机制。但要注意的是分布式锁跟同一个进程的锁还是有区别的，所以使用的时候要比同一个进程里的锁更谨慎的使用

这儿其实说了两个作用

1. 传统意义的锁，如《剖析分布式锁》，保护对共享资料操作
2. master选举，像JOB，为了高可用，会有多台服务器部署同一套JOB程序，但在运行时，只有一台服务器真正执行业务，此时，需要选择一台服务器，如果这台机器挂了，别的机器需要顶替上来

集群管理

在分布式的集群中，经常会由于各种原因，比如硬件故障，软件故障，网络问题，有些节点会进进出出。有新的节点加入进来，也有老的节点退出集群。这个时候，集群中其他机器需要感知到这种变化，然后根据这种变化做出对应的决策

比如我们是一个分布式存储系统，有一个中央控制节点负责存储的分配，当有新的存储进来的时候我们要根据现在集群目前的状态来分配存储节点

这个时候我们就需要动态感知到集群目前的状态，这也就是注册中心

CAP

分布式系统在设计时，都会考虑一下CAP，在现有理论下，CAP是不能同时满足的，所以需要根据业务场景选择合适的设计要求

CAP定义在《zookeeper-paxos》中有详细说明

ZooKeeper是个CP（一致性+分区容错性）的，即任何时刻对ZooKeeper的访问请求能得到一致的数据结果，同时系统对网络分割具备容错性；但是它不能保证每次服务请求的可用性。也就是在极端环境下，ZooKeeper可能会丢弃一些请求，消费者程序需要重新请求才能获得结果。

ZooKeeper是分布式协调服务，它的职责是保证数据在其管辖下的所有服务之间保持同步、一致；所以就不难理解为什么ZooKeeper被设计成CP而不是AP特性的了

而且， 作为ZooKeeper的核心实现算法Zab，就是解决了分布式系统下数据如何在多个服务之间保持同步问题的

特点

• 顺序一致性：从同一客户端发起的事务请求，最终将会严格地按照顺序被应用到 ZooKeeper 中去。
• 原子性：所有事务请求的处理结果在整个集群中所有机器上的应用情况是一致的，也就是说，要么整个集群中所有的机器都成功应用了某一个事务，要么都没有应用。
• 单一系统映像：无论客户端连到哪一个 ZooKeeper 服务器上，其看到的服务端数据模型都是一致的。
• 可靠性：一旦一次更改请求被应用，更改的结果就会被持久化，直到被下一次更改覆盖。

znode

在谈到分布式的时候，我们通常说的“节点”是指组成集群的每一台机器
然而，在 ZooKeeper 中，“节点”分为两类：

1. 第一类同样是指构成集群的机器，我们称之为机器节点。
2. 第二类则是指数据模型中的数据单元，我们称之为数据节点一ZNode

与Linux文件系统不同的是，Linux文件系统有目录和文件的区别，而Zookeeper的数据节点称为ZNode，ZNode是Zookeeper中数据的最小单元，每个ZNode都可以保存数据，同时还可以挂载子节点，因此构成了一个层次化的命名空间，称为树

• 每一个znode默认能够存储1MB的数据（对于记录状态性质的数据来说，够了）

• 可以使用zkCli命令，登录到zookeeper上，并通过ls、create、delete、sync等命令操作这些znode节点

znode除了名称、数据以外，还有一套属性：zxid

ZooKeeper状态的每一次改变, 都对应着一个递增的Transaction id, 该id称为zxid. 由于zxid的递增性质, 如果zxid1小于zxid2, 那么zxid1肯定先于zxid2发生

创建任意节点, 或者更新任意节点的数据, 或者删除任意节点, 都会导致Zookeeper状态发生改变, 从而导致zxid的值增加

此外，znode还有操作权限。如果我们把以上几类属性细化，又可以得到以下属性的细节：

• czxid：创建节点的事务的zxid
• mzxid：对znode最近修改的zxid
• ctime：以距离时间原点(epoch)的毫秒数表示的znode创建时间
• mtime：以距离时间原点(epoch)的毫秒数表示的znode最近修改时间
• version：znode数据的修改次数
• cversion：znode子节点修改次数
• aversion：znode的ACL修改次数
• ephemeralOwner：如果znode是临时节点，则指示节点所有者的会话ID；如果不是临时节点，则为零。
• dataLength：znode数据长度。
• numChildren：znode子节点个数。

znode是由客户端创建的，它和创建它的客户端的内在联系，决定了它的存在性：

1. PERSISTENT-持久化节点：创建这个节点的客户端在与zookeeper服务的连接断开后，这个节点也不会被删除（除非您使用API强制删除）。
2. PERSISTENT_SEQUENTIAL-持久化顺序编号节点：当客户端请求创建这个节点A后，zookeeper会根据parent-znode的zxid状态，为这个A节点编写一个全目录唯一的编号（这个编号只会一直增长）。当客户端与zookeeper服务的连接断开后，这个节点也不会被删除。
3. EPHEMERAL-临时目录节点：创建这个节点的客户端在与zookeeper服务的连接断开后，这个节点（还有涉及到的子节点）就会被删除。
4. EPHEMERAL_SEQUENTIAL-临时顺序编号目录节点：当客户端请求创建这个节点A后，zookeeper会根据parent-znode的zxid状态，为这个A节点编写一个全目录唯一的编号（这个编号只会一直增长）。当创建这个节点的客户端与zookeeper服务的连接断开后，这个节点被删除。

另外，无论是EPHEMERAL还是EPHEMERAL_SEQUENTIAL节点类型，在zookeeper的client异常终止后，节点也会被删除。

服务的四种状态

服务器具有四种状态，分别是LOOKING,FOLLOWING,LEADING,OBSERVING

• LOOKING
  寻找leader状态
  当前服务器处于该状态时，它会认为当前集群中没有leader，因此需要进入leader选举状态
• FOLLOWING
  跟随者状态
  表示当前服务器的角色是Follower角色
• LEADING
  领导者状态
  表示当前服务器是Leader
• OBSERVING
  观察者状态
  表示当前服务器角色是Observer

角色

最典型集群模式：Master/Slave 模式（主备模式）

在这种模式中，通常 Master 服务器作为主服务器提供写服务，其他的 Slave 服务器从服务器通过异步复制的方式获取 Master 服务器最新的数据提供读服务

zookeeper都是集群形式部署的，而zk服务又分为不同角色来执行不同的任务,ZooKeeper中没有选择传统的 Master/Slave 概念

而是引入了Leader、Follower 和 Observer 三种角色

在区分zk服务器角色之前，需要解释几个概念：

• 事务请求
  在zk中，那些会改变服务器状态的请求称为事务请求（创建节点、更新数据、删除节点、创建会话等等）
• 非事务请求
  从zk读取数据但是不对状态进行任何修改的请求称为非事务请求

领导者Leader

1. 事务请求的唯一调度和处理者，保证集群事务处理的顺序性；
2. 集群内部各服务器的调度者
3. 只有一个

跟随者（Follower）

1. 处理客户端非事务请求，转发事务请求给Leader服务器
2. 参与事务请求Proposal的投票
3. 参与Leader选举的投票

观察者（Observer):

1. Follower 和 Observer 唯一的区别在于 Observer 机器不参与 Leader 的选举过程
2. 也不参与写操作的“过半写成功”策略，因此 Observer
3. 机器可以在不影响写性能的情况下提升集群的读性能

数据流

1. 在Client向Follwer发出一个写的请求
2. Follwer把请求发送给Leader
3. Leader接收到以后开始发起投票并通知Follwer进行投票
4. Follwer把投票结果发送给Leader
5. Leader将结果汇总后如果需要写入，则开始写入同时把写入操作通知给Leader，然后commit;
6. Follwer把请求结果返回给Client

leader选举

选举(election)是分布式系统实践中常见的问题，通过打破节点间的对等关系，选得的leader(或叫master、coordinator)有助于实现事务原子性、提升决议效率

为什么需要选举

集群本身有很多种类，如tomcat集群，集群里面每一台机器是对等的，所以其自身不存在leader之说

另外一类，如fastDfs，其依赖于独特的HASH算法，建立文件名和路径之间的映射关系，写操作都是通过namenode分发到各台datanode之上，算法保证了文件名的独一无二，也不存在leader的说法

还有memcache集群，集群里面的机器之间彼此无心跳，通过一致性hash尽可能将key值的存储分散化，降低单一memcahe服务器down机的影响。

还有一类是主从复制，主节点负责写，从节点负责读，提高读的性能。从节点定期通过心跳与主节点沟通，一旦主节点挂掉了，从节点马上接手主节点的任务

对于分布式应用，难以避免出现网络的抖动。比如，
主节点暂时失去响应，如瞬时负载过高，网络拥塞或者其他原因导致主节点暂时失去响应，超过响应超时时间，这个时候从节点启动，承担起leader的职责，但是原先的主节点又恢复了服务。这个时候，如果没有选举机制（不能仅仅自己宣告自己是leader，还要广而告之，让其他服务器或者客户端知道自己是leader），有可能会存在两个leader节点，导致集群发生混乱

上图示例一下此类场景

主节点出现问题，那就是单点故障

传统方式是采用一个备用节点，这个备用节点定期给当前主节点发送ping包，主节点收到ping包以后向备用节点发送回复Ack，当备用节点收到回复的时候就会认为当前主节点还活着，让他继续提供服务

当主节点挂了，这时候备用节点收不到回复了，然后他就认为主节点挂了接替他成为主节点如下图

但是这种方式就是有一个隐患，就是网络问题，来看一网络问题会造成什么后果，如下图

也就是说我们的主节点的并没有挂，只是在回复的时候网络发生故障，这样我们的备用节点同样收不到回复，就会认为主节点挂了，然后备用节点将他的Master实例启动起来，

这样我们的分布式系统当中就有了两个主节点也就是—双Master，

出现Master以后我们的从节点就会将它所做的事一部分汇报给了主节点，一部分汇报给了从节点，这样服务就全乱了

选举

场景

在哪些场景下需要进行leader选举

1. 服务器初始化启动
2. 服务器运行期间无法和Leader保持连接

初始化时

若进行Leader选举，则至少需要两台机器，这里选取3台机器组成的服务器集群为例。在集群初始化阶段，当有一台服务器Server1启动时，其单独无法进行和完成Leader选举，当第二台服务器Server2启动时，此时两台机器可以相互通信，每台机器都试图找到Leader，于是进入Leader选举过程。选举过程如下

• 1.每个Server发出一个投票。由于是初始情况，Server1和Server2都会将自己作为Leader服务器来进行投票，每次投票会包含所推举的服务器的myid和ZXID，使用(myid, ZXID)来表示，此时Server1的投票为(1, 0)，Server2的投票为(2, 0)，然后各自将这个投票发给集群中其他机器。
• 2.接受来自各个服务器的投票。集群的每个服务器收到投票后，首先判断该投票的有效性，如检查是否是本轮投票、是否来自LOOKING状态的服务器
• 3.处理投票。针对每一个投票，服务器都需要将别人的投票和自己的投票进行PK，PK规则如下
  • 3.1.优先检查ZXID。ZXID比较大的服务器优先作为Leader
  • 3.2.如果ZXID相同，那么就比较myid。myid较大的服务器作为Leader服务器

对于Server1而言，它的投票是(1, 0)，接收Server2的投票为(2, 0)，首先会比较两者的ZXID，均为0，再比较myid，此时Server2的myid最大，于是更新自己的投票为(2, 0)，然后重新投票，对于Server2而言，其无须更新自己的投票，只是再次向集群中所有机器发出上一次投票信息即可。

• 4.统计投票。每次投票后，服务器都会统计投票信息，判断是否已经有过半机器接受到相同的投票信息，对于Server1、Server2而言，都统计出集群中已经有两台机器接受了(2, 0)的投票信息，此时便认为已经选出了Leader。
• 5.改变服务器状态。一旦确定了Leader，每个服务器就会更新自己的状态，如果是Follower，那么就变更为FOLLOWING，如果是Leader，就变更为LEADING。

Leader挂掉

在Zookeeper运行期间，Leader与非Leader服务器各司其职，即便当有非Leader服务器宕机或新加入，此时也不会影响Leader，但是一旦Leader服务器挂了，那么整个集群将暂停对外服务，进入新一轮Leader选举，其过程和启动时期的Leader选举过程基本一致。假设正在运行的有Server1、Server2、Server3三台服务器，当前Leader是Server2，若某一时刻Leader挂了，此时便开始Leader选举。选举过程如下

• 1.变更状态。Leader挂后，余下的非Observer服务器都会讲自己的服务器状态变更为LOOKING，然后开始进入Leader选举过程
• 2.每个Server会发出一个投票。在运行期间，每个服务器上的ZXID可能不同，此时假定Server1的ZXID为123，Server3的ZXID为122；在第一轮投票中，Server1和Server3都会投自己，产生投票(1, 123)，(3, 122)，然后各自将投票发送给集群中所有机器。
• 3.接收来自各个服务器的投票。与启动时过程相同。
• 4.处理投票。与启动时过程相同，此时，Server1将会成为Leader。
• 5.统计投票。与启动时过程相同。
• 6.改变服务器的状态。与启动时过程相同

过程

理想状态

1. 在第一轮中，按照“我最牛逼，我怕谁”的原则，每个节点都推荐它自己为集群的leader节点
2. 按照我们假设的理想条件，节点S1首先收到了S2发送来的推荐者“2”，节点S1发现“2”要比它之前推荐的“1”（也就是它自己）牛。根据谁牛推荐谁的原则，“S1”清空自己的票箱，重新选举“2”（注意，此时“S1”的新票箱中已经有两票选举“2”了，一票是它自己，另外一票是”S2”，并且所有节点都是Looking状态）
3. 同样的事情发生在“S2”身上：”S2”收到了”S3”发过来的推荐信息，发现“3”这个被推举者比之前自己推举的“2”要牛，于是也清空自己的票箱，发起一轮新的投票，此时“S2”选举“3”。依次类推”S3”、”S4”
4. 这里要注意S5这个节点，在第一轮接受到了来源于“S1”——“S4”的推举者（一定注意，每一次接受信息，都会广播一次“我坚持推举的人”），发现“还是推荐的5最牛”，于是“我继续推举S5吧”
5. 以上这个过程在整个理想的网络环境上一直持续。到了第四轮，“S1”收到了“S2”发送来的推举者“5”，发现“5”要比当前“S1”推荐的“4”要牛。所以“S1”清空了自己的票箱，重新推举“5”（发送给其他所有节点）
6. 关键的第五轮来了，我们再重复一下，经过之前的选举，现在“S2”——“S5”都已经推举“5”为Leader了，而且都处于第四轮。这时他们收到了”S1”发来的新的“第五轮”投票，于是都和之前一样，做相同的一件事：清空自己的票箱，重新向其他所有节点广播自己的第五轮投票“5”
7. 于是，节点X，收到了大于N / 2 +１的选举“5”的投票，且都是第五轮投票。这样每个节点就都知道了自己的角色。，选举结束。所有将成为Follower状态的节点，向将要成为Leader的节点发起最后一次“工作是否正常”的询问。得到肯定的ack后，整个集群的工作状态就确认了

非理想状态

过程中出现了宕机、网络延迟、网络物理层断开等情况

在第三轮的选举过程后，“S1”,“S2”两个节点就断开了，他们的投票信息根本没有发送出去

• “S3”收到了“S4”,“S5”发来的投票信息，这时“S3”的票箱处于第3轮，并且发现了占大多数的投票结果：大家选举“S5”为Leader节点
• 同样的事情也发生在“S4”身上。这样“S3”,“S4”两个节点率先知道了投票结果，在最后一次询问Leader节点是否能正常工作，并得到了肯定的ACK之后，“S3”,“S4”两个节点变成了Follower状态
• 之后，无论“S3”,“S4”两个节点收到了任何节点的投票信息，都直接向源节点反馈投票结果，不会再进行投票了。
• 在投票完成后，“S1”,“S2”重新连入后，虽然他们发起了投票，但是不会再收到投票反馈了。直接根据“S3”或者“S4”发来的结果状态，变成Follower状态

总结

此篇zookeeper的基础内容基本都包含了

了解这些基本已经入门

下一篇学习其中最核心的zab协议，理解原子性广播概念，以及zab的实现过程

参考资料

面试问题，请说明zookeeper的选举机制

zookeeper选举机制

Zookeeper的Leader选举

再起航系列,一枚互联网菜鸟的成长历程

春风三月，却是很多职场人的混沌时节

总结性文章，让人难以落笔。可能观察生活、体察自己不够细心；也许是成长缓慢，后知后觉；更可能本身就是平庸，普通人平平淡淡，没有什么轰烈可期的事情。但不管多难，终需落笔，无论什么原因，总得回顾总结

来到互联网行业已经两年，在这个时间节点，可能再次处于混沌

在第一年里，从好奇，新鲜，无知，疑惑 到 熟悉，充足；再到第二年，可能就是重复，机械，麻木，厌烦了。

在刚换行时，一切都是新鲜，好奇的。经过一年时间的积累，对业务已经熟悉，能够应对各种业务需求，并能很好地实践业务场景中的技术栈，自己的能力圈在扩大，给人一种很充实的感觉

慢慢地，从刚开始还能对技术栈的细节，深度保持一颗期待之心，陷入了各种业务需求中，毕竟是要解决问题体现价值的，不能仅仅学习新鲜事物，这也是前辈们常讲的不要抱着进大公司学习技术的心态，热爱学习的人在哪儿都能学习，并且学好

你也是这么被洗脑的，因此你熟悉了各种业务场景，处理各样的业务咨询以及各环境的故障，帮上下游解决问题，开始重复生活，机械操作，厌烦了，总会厌烦，也有能力厌烦，毕竟你支撑了团队，混熟了业务线上的同事，老油条诞生了，大公司的螺丝钉也铸成了

这些感受都是来自外界的刺激，似乎是常理的周期反应，此时需要一丝保持平静，进步的心，都会向外寻找答案，寻求机会。一切都得向外扩张

但不管如何向外，都会落空，不踏实，终需回归自我，向内而问。不忘初心，方得始终

从工作之初，有一种思想一直占据内心，那就是跟随团队，公司成长，是最夯实，也是最快的成长试。奈何，在实现中，几任老东家都走向了末路，而我也付之东流。尤其这次，完全换了行业，一切从零开始，命也，运也？

以前厌倦了项目总是失败，总是从头开始，总想着能有一个成功的项目，不停地迭代，走向完美。这一次终于满足了，就一个项目不停地迭代，让你迭代个够。每一次迭代，历史包袱特别大，想想都替命运着急，过去走在创新的路上，嫌弃没有积累，现在迭代，又抱怨历史包袱了，人真难伺候

混沌期总是迷茫，没有方向。牛人总有规划，人生规划。普通人总是在寻找方向，选择方向，而又没有方向

在《再起航三转行互联网》中，提到程序员分为两类：技术型和产品型

技术型，一直在路上，只要没有脱离技术型公司，这永远在路上，技术人不可能，也不能丢弃技术。不管老酒装新瓶，还是新酒装老瓶，都是保持与技术的距离。但也得认清自己，不可能做到技术大牛，技术领跑者。智商，经历沉淀都已经让我无法走上这条路

产品型，但你能跨越三十五岁魔咒吗？体力，精力方面是得向年轻人低头的。

那前途在何方？哪片土地是你希望的田野？

管理领导？但现在不在其位置，考虑学习太多理论，都是形而上学。终归要落实到形而下学上

路在何方呢？

何时拨开云雾，走出混沌！

可能需要未来的你给出答案

现在可以没有答案，甚至说未来也没有答案，但行动不能停止，思考更不能停止

从第二年的开始，已经开始关注速度《再起航二成长速度》，第一年得以业务为核心，了解熟悉业务，把手头工作处理好。第二年就得把在第一年关注好奇的技术进行深入研究了，所以开始以组件为单位，一个一个了解学习，以微服务为基架，深入每一个环节

是的，来大公司不是学习的，老板聘用你，不是让你来学习的，但自己得知道，学习一刻都不能停，到大公司，的确不是学习的，在哪儿都能学习，主要是有一个场景，一个能更加深入层次学习的场景，好比之前的《剖析分布式锁》，在一般公司错误的写法，准确的说，不严谨的实现也能胜任，但更大业务量时，能不能胜任，需要思考，更多的时候，需要一个场景告诉你，你当下的完美，可能是个大大的残次品

现在不管是外部大环境，还是公司内部小环境，都不理想，尤其公司现在的体量，是不需要我的陪伴成长的，所以前方还有别的路，创业别想，成熟条件完全没有，那只能去一家与自己相匹配的公司，一起成长

公司不能大，也不能小。中型有成长的公司，核心业务部门，深入业务；或者基础架构，让面试造火箭，名副其实

可能现在格局太小，总是以技术为一切；此篇只是总结当前的一种状态，不抱怨，不迷信，何时开悟走出混沌，尽人事，知天命，一切随缘

从第一篇《算法概要》开始，到此篇已经经历了将近四个月时间，常见的基础排序已经温习完成

内外排序

内部排序：待排序记录存放在计算机随机存储器中（说简单点，就是内存）进行的排序过程

外部排序：待排序记录的数量很大，以致于内存不能一次容纳全部记录，所以在排序过程中需要对外存进行访问的排序过程

衡量效率

内部排序：比较次数，也就是时间复杂度

外部排序：IO次数，也就是读写外存的次数

IO对排序的影响可以阅读《深入浅出索引》体会

算法

详细介绍

算法渣-排序-冒泡

冒泡排序，应该是很多人会且只会的算法；两两比较交换

为了减小比较与交换的次数，通过双向比较(鸡尾酒排序)、设定是否交换位实现

算法渣-排序-快速排序

快速排序，相对冒泡又改进了，都是交换，但引入了分治思想

算法渣-排序-插入

插入排序，像打牌时，整理牌一样，通过比较、移动来达到排序

算法渣-排序-希尔

希尔，相对插入做了改进，不是一步一步的移动，而是大步大步的移动

算法渣-排序-选择

选择，类似插入的反向操作；插入排序是边读边排，每当读入一个新的数时，目前的数组一定是排好序的。而选择排序不同，它必须是读完所有的数据之后才能开始排序的

算法渣-排序-堆排序

堆排序，借助堆数据结构，构造堆结构，选取堆顶元素，不再需要遍历所有元素选择

算法渣-排序-归并排序

归并排序，也是分治思想，但与快速有些区别；归并排序是由下向上的，先处理子数组然后再合并。而快速排序正好相反，它的过程是由上向下的，先分出两个子区间，再对子区间进行排序

算法渣-排序-基数排序

算法渣-排序-桶排序

算法渣-排序-计数排序

线性排序算法,非基于比较的排序算法，性能很高，但都有限制才能达到线性排序的效果

场景

对于排序算法选择，不能单从时间复杂上看，简单算法都是O(n^2),就不考虑，只选择改进算法

插入排序 vs 快速排序 vs 归并排序

由下图可以看出，在输入规模小于100时，插入排序要好于归并和快速排序。在输入规模小于200时，插入排序优于归并排序。规模在30以下时，插入排序效率要比快速排序高50%以上，规模在50以下时，插入排序比归并排序效率高90%以上

改进算法

在数据量大时，使用改进算法

1. 就时间性能而言， 希尔排序、快速排序、树形选择排序、堆排序和归并排序都是较为先进的排序方法。耗时远小于O(N^2)级别的算法。
2. 先进算法之中，快排的效率是最高的。 但其缺点十分明显：在待排序列基本有序的情况下，会蜕化成起泡排序，时间复杂度接近 O(N^2)。
3. 希尔排序的性能让人有点意外，这种增量插入排序的高效性完全说明了：在基本有序序列中，直接插入排序绝对能达到令人吃惊的效率。但是希尔排序对增量的选择标准依然没有较为满意的答案，要知道增量的选取直接影响排序的效率。
4. 归并排序的效率非常不错，在数据规模较大的情况下，它比希尔排序和堆排序都要好。
5. 堆排序在数据规模较小的情况下还是表现不错的，但是随着规模的增大，时间代价也开始和上面两种排序拉开的距离。

总结

总的来说，并不存在“最佳”的排序算法。必须针对待排序列自身的特点来选择“良好”的算法。下面有一些指导性的意见：

1. 数据规模很小，而且待排序列基本有序的情况下，选择直接插入排序绝对是上策。不要小看它O(N^2)级别
2. 数据规模不是很大，完全可以使用内存空间。而且待排序列杂乱无序(越乱越开心)，快排永远是不错的选择，当然付出log(N)的额外空间是值得的。
3. 海量级别的数据，必须按块存放在外存(磁盘)中。此时的归并排序是一个比较优秀的算法

试题

【京东】假设你只有100Mb的内存，需要对1Gb的数据进行排序，最合适的算法是（ ）

A. 归并排序　　B. 插入排序　　C. 快速排序　　D. 冒泡排序

根据题目，我们可以知道，我们现有的内存限制使得我们无法把数据一次性加载到内存中，所以我们只能先加载一部分数据，对其排序后存入磁盘中。然后再加载一些数据，把它们“合并”到已排序的数据集中去，重复这个过程直到排序完成，显然最能胜任这个工作的是归并排序。

【2016阿里巴巴校招笔试题】现有1GB数据进行排序，计算资源只有1GB内存可用，下列排序方法中最可能出现性能问题的是（ ）

A. 堆排序　　B. 插入排序　　C. 归并排序　　D. 快速排序　　E. 选择排序　　F. 冒泡排序

根据题目的描述，我们能够很明确的知道这道题考察我们的是原地排序的概念，这里我们只需要选择非原地排序的占用额外空间最大的算法，显然答案是”C. 归并排序”。

【2015阿里巴巴研发工程师笔试题】个数约为50K的数列需要进行从小到大排序，数列特征是基本逆序(多数数字从大大小，个别乱序)，以下哪种排序算法在事先不了解数列特征的情况下性能最优

A. 冒泡排序　　B. 改进冒泡排序　　C. 选择排序　　D. 快速排序　　E. 堆排序　　F.插入排序

根据题目中的描述，首先我们可以排除A、B、C，因为它们的时间复杂度都是O(n^2)。接下来我们看下D选项，我们前面提到过，快速排序在最坏情况下的时间复杂度会退化至O(n^2)，F选项的插入排序在逆序数很大时性能也很差（O(n^2)）。而堆排序在最坏情况下的复杂度也为O(logn)，所以这里我们应该选择堆排序

参考资料

基于比较的内部排序总结

常见比较排序算法的耗时测试

背景

图片分库存储时，每一张图片都可以定位到特定的服务器

hash

上图中，假设我们查找的是”a.png”，由于有4台服务器（排除从库），因此公式为hash(a.png) % 4 = 2 ，可知定位到了第2号服务器，这样的话就不会遍历所有的服务器，大大提升了性能！

一切都运行正常，再考虑如下的两种情况；

1. 一个 cache 服务器 m down 掉了（在实际应用中必须要考虑这种情况），这样所有映射到 cache m 的对象都会失效，怎么办，需要把 cache m 从 cache 中移除，这时候 cache 是 N-1 台，映射公式变成了 hash(object)%(N-1) ；
2. 由于访问加重，需要添加 cache ，这时候 cache 是 N+1 台，映射公式变成了 hash(object)%(N+1) ；

1 和 2 意味着什么？

这意味着突然之间几乎所有cache 都失效了。缓存雪崩，这是一场灾难

一致性hash

有什么方法可以改变这个状况呢，这就是 consistent hashing

比如有{N0, N1, N2}三个节点，陆续有多个资源要分配到这三个节点上，如何尽可能均匀的分配到这些节点上

算法

一致性哈希算法的思路为：先构造出一个长度为2^32 整数环，根据N0-3的节点名称的hash值（分布为[0,2^32 -1]）放到这个环上

整个空间按顺时针方向组织，圆环的正上方的点代表0，0点右侧的第一个点代表1，以此类推，2、3、4、5、6……直到2^32 -1，也就是说0点左侧的第一个点代表2^32 -1， 0和2^32 -1在零点中方向重合，我们把这个由2^32个点组成的圆环称为Hash环

下一步将各个服务器使用Hash进行一个哈希，具体可以选择服务器的ip或主机名作为关键字进行哈希，这样每台机器就能确定其在哈希环上的位置，这里假设四台服务器使用ip地址哈希后在环空间的位置如下：

接下来使用如下算法定位数据访问到相应服务器：将数据key使用相同的函数Hash计算出哈希值，并确定此数据在环上的位置，从此位置沿环顺时针“行走”，第一台遇到的服务器就是其应该定位到的服务器！

例如我们有Object A、Object B、Object C、Object D四个数据对象，经过哈希计算后，在环空间上的位置如下：

根据一致性Hash算法，数据A会被定为到Node A上，B被定为到Node B上，C被定为到Node C上，D被定为到Node D上

容错性

如果一个节点宕机了，会引起系统故障吗？

如上图，Node C不幸宕机，可以看到此时对象A、B、D不会受到影响，只有C对象被重定位到Node D。一般的，在一致性Hash算法中，如果一台服务器不可用，则受影响的数据仅仅是此服务器到其环空间中前一台服务器（即沿着逆时针方向行走遇到的第一台服务器）之间数据，其它不会受到影响

扩展性

如果在系统中增加一台服务器Node X

此时对象Object A、B、D不受影响，只有对象C需要重定位到新的Node X ！一般的，在一致性Hash算法中，如果增加一台服务器，则受影响的数据仅仅是新服务器到其环空间中前一台服务器（即沿着逆时针方向行走遇到的第一台服务器）之间数据，其它数据也不会受到影响

综上所述，一致性Hash算法对于节点的增减都只需重定位环空间中的一小部分数据，具有较好的容错性和可扩展性。

数据倾斜

一致性Hash算法在服务节点太少时，容易因为节点分部不均匀而造成数据倾斜（被缓存的对象大部分集中缓存在某一台服务器上）问题，例如系统中只有两台服务器，其环分布如下：

此时必然造成大量数据集中到Node A上，而只有极少量会定位到Node B上

然而，这又会造成一个“雪崩”的情况，即A节点由于承担了B节点的数据，所以A节点的负载会变高，A节点很容易也宕机，这样依次下去，这样造成整个集群都挂了

虚拟节点

计算机的任何问题都可以通过增加一个虚拟层来解决

解决上述数据倾斜问题，也可能通过使用虚拟层的手段：将每台物理缓存服务器虚拟为一组虚拟缓存服务器，将虚拟服务器的hash值放置在hash环上，Key在环上先找到虚拟服务器节点，再得到物理服务器的信息

例如上面的情况，可以为每台服务器计算三个虚拟节点，于是可以分别计算 “Node A#1”、“Node A#2”、“Node A#3”、“Node B#1”、“Node B#2”、“Node B#3”的哈希值，于是形成六个虚拟节点：

同时数据定位算法不变，只是多了一步虚拟节点到实际节点的映射，例如定位到“Node A#1”、“Node A#2”、“Node A#3”三个虚拟节点的数据均定位到Node A上。这样就解决了服务节点少时数据倾斜的问题

那么在实践中，一台物理服务器虚拟为多少个虚拟服务器节点合适呢？太多会影响性能，太少又会导致负载不均衡，一般说来，经验值是150，当然根据集群规模和负载均衡的精度需求，这个值应该根据具体情况具体对待

实现

判定哈希算法好坏的四个定义：

1. 平衡性(Balance)：平衡性是指哈希的结果能够尽可能分布到所有的缓冲中去，这样可以使得所有的缓冲空间都得到利用。很多哈希算法都能够满足这一条件
2. 单调性(Monotonicity)：单调性是指如果已经有一些内容通过哈希分派到了相应的缓冲中，又有新的缓冲加入到系统中。哈希的结果应能够保证原有已分配的内容可以被映射到原有的或者新的缓冲中去，而不会被映射到旧的缓冲集合中的其他缓冲区
3. 分散性(Spread)：在分布式环境中，终端有可能看不到所有的缓冲，而是只能看到其中的一部分。当终端希望通过哈希过程将内容映射到缓冲上时，由于不同终端所见的缓冲范围有可能不同，从而导致哈希的结果不一致，最终的结果是相同的内容被不同的终端映射到不同的缓冲区中。这种情况显然是应该避免的，因为它导致相同内容被存储到不同缓冲中去，降低了系统存储的效率。分散性的定义就是上述情况发生的严重程度。好的哈希算法应能够尽量避免不一致的情况发生，也就是尽量降低分散性
4. 负载(Load)：负载问题实际上是从另一个角度看待分散性问题。既然不同的终端可能将相同的内容映射到不同的缓冲区中，那么对于一个特定的缓冲区而言，也可能被不同的用户映射为不同 的内容。与分散性一样，这种情况也是应当避免的，因此好的哈希算法应能够尽量降低缓冲的负荷

在具体实现时，主要考虑点选择适合的数据结构构造hash环

此数据结构的特点：插入与删除性能、快速找到特定元素的下一位

常见算法结构可以有回顾：

《一篇解决排序算法》

《树结构概述》

从时间复杂度方面选择，使用平衡二叉树数据结构，可以使得查找的时间复杂度降低为O(logN)

使用java，以TreeMap为例，TreeMap本身还提供了一个tailMap(K fromKey)方法，支持从红黑树中查找比fromKey大的值的集合，但并不需要遍历整个数据结构

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

import java.util.Collection;
import java.util.SortedMap;
import java.util.TreeMap;

public class ConsistentHash<T> {

 private final HashFunction hashFunction;
 private final int numberOfReplicas;
 private final SortedMap<Integer, T> circle = new TreeMap<Integer, T>();

 public ConsistentHash(HashFunction hashFunction, int numberOfReplicas,
     Collection<T> nodes) {
   this.hashFunction = hashFunction;
   this.numberOfReplicas = numberOfReplicas;

   for (T node : nodes) {
     add(node);
   }
 }

 public void add(T node) {
   for (int i = 0; i < numberOfReplicas; i++) {
     circle.put(hashFunction.hash(node.toString() + i), node);
   }
 }

 public void remove(T node) {
   for (int i = 0; i < numberOfReplicas; i++) {
     circle.remove(hashFunction.hash(node.toString() + i));
   }
 }

 public T get(Object key) {
   if (circle.isEmpty()) {
     return null;
   }
   int hash = hashFunction.hash(key);
   if (!circle.containsKey(hash)) {
     SortedMap<Integer, T> tailMap = circle.tailMap(hash);
     hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
   }
   return circle.get(hash);
 }
}

dubbo实现

理解完理论，再扒一下工业级产品的运用，dubbo负载均衡策略之一ConsistentHashLoadBalance

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

private final TreeMap<Long, Invoker<T>> virtualInvokers;

ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
    this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
    this.identityHashCode = identityHashCode;
    URL url = invokers.get(0).getUrl();
    this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
    String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
    argumentIndex = new int[index.length];
    for (int i = 0; i < index.length; i++) {
        argumentIndex[i] = Integer.parseInt(index[i]);
    }
    for (Invoker<T> invoker : invokers) {
        String address = invoker.getUrl().getAddress();
        for (int i = 0; i < replicaNumber / 4; i++) {
            byte[] digest = md5(address + i);
            for (int h = 0; h < 4; h++) {
                long m = hash(digest, h);
                virtualInvokers.put(m, invoker);
            }
        }
    }
}

整体思想是一样的，虚拟节点+TreeMap,但实现得更精致，使用MD5加密KEY，更加平衡性，整体的思路解释：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

//对所有节点，生成nCopies个虚拟结点  
for(Node node : nodes) {  
    //每四个虚拟结点为一组，为什么这样？下面会说到  
    for(int i=0; i<nCopies / 4; i++) {  
        //getKeyForNode方法为这组虚拟结点得到惟一名称  
        byte[] digest=HashAlgorithm.computeMd5(getKeyForNode(node, i));  
    /** Md5是一个16字节长度的数组，将16字节的数组每四个字节一组，分别对应一个虚拟结点，这就是为什么上面把虚拟结点四个划分一组的原因*/  
        for(int h=0;h<4;h++) {  
          //对于每四个字节，组成一个long值数值，做为这个虚拟节点的在环中的惟一key
          //结果转换为long类，这是因为生成的结果是一个32位数，若用int保存可能会产生负数。而一致性hash生成的逻辑环其hashCode的范围是在 0 - MAX_VALUE之间。因此为正整数，所以这里要强制转换为long类型，避免出现负数。
            Long k = ((long)(digest[3+h*4]&0xFF) << 24)  
                | ((long)(digest[2+h*4]&0xFF) << 16)  
                | ((long)(digest[1+h*4]&0xFF) << 8)  
                | (digest[h*4]&0xFF);  
            allNodes.put(k, node);  
        }  
    }  
}

总结

QA

为什么hash一致性的数据空间范围是2^32次方？

这个问题有两种答案，一是技术限制、一是实际场景：

1. 因为，java中int的最大值是2^31-1最小值是-2^31,2^32刚好是无符号整形的最大值
2. 因为一致性hash算法是来做服务器的负载均衡，而服务器的IP地址是32位，所以是2^32-1次方的数值空间

进一步追尾基础，为什么java中int的最大值是2^31-1最小值是-2^31？

因为，int的最大值最小值范围设定是因为一个int占4个字节，一个字节占8位，二进制中刚好是32位

根据算法特性，一致性hash是最好的选择吗？

下一篇介绍另一种实现google maglev hashing算法

参考资料

《大型网站技术架构》

对一致性Hash算法，Java代码实现的深入研究

为什么hash环是32位

没有一身好内功，招式再多都是空;算法绝对是防身必备，面试时更是不可或缺；跟着算法渣一起从零学算法

线性排序

常见的三种以线性时间运行的算法：计数排序、基数排序和桶排序

需要注意的是线性排序算法是非基于比较的排序算法，都有使用限制才能达到线性排序的效果

线性排序是个神奇的算法，比基数排序及桶排序神奇得多

定义

计数排序是一个非基于比较的排序算法，该算法于1954年由 Harold H. Seward 提出。它的优势在于在对一定范围内的整数排序时，它的复杂度为Ο(n+k)（其中k是整数的范围），快于任何比较排序算法

算法

计数排序的基本思想是对于给定的输入序列中的每一个元素x，确定该序列中值小于x的元素的个数（此处并非比较各元素的大小，而是通过对元素值的计数和计数值的累加来确定）。一旦有了这个信息，就可以将x直接存放到最终的输出序列的正确位置上

首先需要三个数组，第一个数组记录A要排序的数列大小为n，第二个数组B要记录比某个数小的其他数字的个数所以第二个数组的大小应当为K（数列中最大数的大小），第三个数组C为记录排序好了的数列的数组，大小应当为n。

接着需要确定数组最大值并确定B数组的大小。并对每个数由小到大的记录数列中每个数的出现次数。因为是有小到大通过出现次数可以通过前面的所有数的出现次数来确定比这个数小的数的个数，从而确定其位置

假定20个随机整数的值，也就是第一个数组A：

9，3，5，4，9，1，2，7，8，1，3，6，5，3，4，0，10，9 ，7，9

数组A中最大数为10，数组B需要0~10索引

比如第一个整数是9，那么数组下标为9的元素加1：

第二个整数是3，那么数组下标为3的元素加1：

最终，数列遍历完毕时，数组的状态如下：

数组每一个下标位置的值，代表了数列中对应整数出现的次数。

有了这个“统计结果”，排序就很简单了。直接遍历数组，输出数组元素的下标值，元素的值是几，就输出几次：

0，1，1，2，3，3，3，4，4，5，5，6，7，7，8，9，9，9，9，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
29
30
31
32
33
34
35
36
37
38
39
40

/**
 * 计数排序
 * @param array
 */
public static void countSort(int []array){
    //最大最小数
    int min = Integer.MAX_VALUE;
    int max = Integer.MIN_VALUE;
    for ( int a:array) {
        if(min > a) {
            min = a;
        }
        if(max < a) {
            max = a;
        }
    }

    //数组索引数量,统计数组计数
    // max-min 主要是考虑到像[90,99,93,95]不是从很小数开始的数组排序，减小空间消耗
    int indexCount = max - min + 1;
    System.err.println("统计数组长度"+indexCount);
    int []countArray = new int[indexCount];

    for (int i=0;i<array.length;i++){
        System.err.println(array[i]-min);
        countArray[array[i]-min]++;
    }
    System.err.println("countArray:"+Arrays.toString(countArray));

    //排好序的数组
    int [] sortArray = new int[array.length];
    int index = 0;
    for (int i=0;i<indexCount;i++) {
        for (int j=0;j<countArray[i];j++){
            sortArray[index++] = min + i;
        }
    }
    //输出就是有序
    System.err.println(Arrays.toString(sortArray));
}

改进

上面的实现为什么要改进，主要是当两个元素相同时，算法的稳定性问题

改进之前，数组B存放的是元素出现的次数

改进之后，会引入一个新数组(也可以共用数组B)，存放的是元素的位置号(这个纯粹是个人理解，之前数组B是存放元素出现的次数，新数组存放每一个元素都加上前面所有元素出现次数之和，当找到对应排序数组元素时，新数组元素就是位置号)

语言比较空洞，直接来个示例(转自小灰程序员)

将数组arr中的数据当作是学生的成绩，要求不但要按照顺序从低到高排序，成绩相同时，按原有顺序显示：

统计数组从第二个元素开始，每一个元素都加上前面所有元素之和

第一步，我们遍历成绩表最后一行的小绿：

小绿是95分，我们找到countArray下标是5的元素，值是4，代表小绿的成绩排名位置在第4位。

同时，我们给countArray下标是5的元素值减1，从4变成3,，代表着下次再遇到95分的成绩时，最终排名是第3。

第二步，我们遍历成绩表倒数第二行的小白：

小白是94分，我们找到countArray下标是4的元素，值是2，代表小白的成绩排名位置在第2位。

同时，我们给countArray下标是4的元素值减1，从2变成1,，代表着下次再遇到94分的成绩时（实际上已经遇不到了），最终排名是第1。

第三步，我们遍历成绩表倒数第三行的小红：

小红是95分，我们找到countArray下标是5的元素，值是3（最初是4，减1变成了3），代表小红的成绩排名位置在第3位。

同时，我们给countArray下标是5的元素值减1，从3变成2,，代表着下次再遇到95分的成绩时（实际上已经遇不到了），最终排名是第2。

这样一来，同样是95分的小红和小绿就能够清楚地排出顺序了，也正因此，优化版本的计数排序属于稳定排序。

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

/**
 * 稳定计数排序
 * @param array
 */
public static void countSort1(int []array) {
    //最大最小数
    int min = Integer.MAX_VALUE;
    int max = Integer.MIN_VALUE;
    for ( int a:array) {
        if(min > a) {
            min = a;
        }
        if(max < a) {
            max = a;
        }
    }

    //数组索引数量,统计数组计数
    // max-min 主要是考虑到像[90,99,93,95]不是从很小数开始的数组排序，减小空间消耗
    int indexCount = max - min + 1;
    System.err.println("统计数组长度"+indexCount);
    int []countArray = new int[indexCount];

    for (int i=0;i<array.length;i++){
        System.err.println(array[i]-min);
        countArray[array[i]-min]++;
    }
    System.err.println("countArray:"+Arrays.toString(countArray));
    //位置数组
    int []pointArray = new int[indexCount];

    int sum =0;
    for (int i = 0;i<indexCount;i++){
        sum += countArray[i];
        pointArray[i] = sum;
    }
    System.err.println("pointArray:"+Arrays.toString(pointArray));
    System.err.println("aaaaaArray:"+Arrays.toString(array));
    //排好序的数组
    int [] sortArray = new int[array.length];
    for (int i=array.length-1;i>=0;i--) {
            sortArray[pointArray[array[i]-min] -1 ] = array[i];
            pointArray[array[i]-min]--;
    }
    //输出就是有序
    System.err.println(Arrays.toString(sortArray));
}

总结

复杂度

假设array元素有N个，取值范围是M。

时间复杂度：3N(计算最大最小数、计数、排好序的数组) + M(位置数组)，去掉系数，时间复杂度是O(N+M)

空间复杂度：只考虑统计数组，那就是M

局限性

1.当数列最大最小值差距过大时，并不适用计数排序。

比如给定20个随机整数，范围在0到1亿之间，这时候如果使用计数排序，需要创建长度1亿的数组。不但严重浪费空间，而且时间复杂度也随之升高。

2.当数列元素不是整数，并不适用计数排序。

如果数列中的元素都是小数，比如25.213，或是0.00000001这样子，则无法创建对应的统计数组。这样显然无法进行计数排序。

引申阅读

算法渣-排序-基数排序

算法渣-排序-桶排序

参考资料

漫画：什么是计数排序

1. 为什么需要限流
2. 如何限流

限流主要就是考虑这两点

为什么需要限流

之前已经介绍了熔断，降级，为什么还需要一个限流呢？是不是多此一举呢？

先来个结论：

熔断是为了防止雪崩，明哲保身；而限流则是在已有条件下，最大限制发挥系统效能

根据《熔断机制》可以知道，熔断有三种状态，[熔断关闭],[半熔断],[熔断开启]三种状态，如果系统压力过大，一个服务就会在三种状态来回切换

会出现一种情况，就像一辆开在崎岖山路上的法拉利，不管车的性能多好，都需要不停的加速减速

要想速度达到最佳，就得让车开在一条笔直的高速公路上

系统就是一条河，服务就像行驶在河里的船，岸的两边，一边是熔断，另一边就是限流；一个保障系统安全，一个保持最大限度运转，让系统达到高可用

如何限流

限流如何实施？

1. 量化限流阀值
2. 确定限流策略、算法
3. 被限制流量的处理

限流阀值，这个其实就是通过系统压力测试来确定

这个工作其实在系统开发之初就需要有初步的估量，涉及到业务规模，增长速度，架构选择等等，根据现有资源及其服务能力，给出上限值

在《计数器算法》中已经说明了几种限流算法：固定窗口、滑动窗口、漏桶、令牌桶

有人总结为【两窗两桶】，很形象

固定窗口：临界问题，一旦流量波动，计数器提前计满，剩余时间都会被限流

滑动窗口：固定窗口的「固定周期」已经很小，那么使用滑动窗口也就没有意义，虽然大大减小临界问题，但总归需要预估一个窗口量，很难把握，过小引起性能和资源损耗

漏桶：不管进来多少量，出去的速率是恒定的，“出口”速度固定；当桶容量满时，就会被限流

令牌桶：“进口”速度固定；只要令牌的生成速度大于等于请求被处理的速度,系统就能正常处理

有了限流策略，那制定在服务端还是客户端呢？

从效果上讲，肯定是客户端，限制越靠入口，系统就越安全，可以防止不必要的无效流量进入服务端系统

你也可能想到对网络资源的浪费，但一般各个服务都是在同一IDC机房，因此这点不必考虑

但成本相对在服务端过高，放在客户端，客户端的数量与扩张速度也是动态的，无法评估，对每个端的限流阀值分配与下发都很复杂

因此一般都是在服务端进行限流

至于被限制的流量如何处理？大多数情况是直接抛弃，在系统之初，就有了对流量预判，从设计到资源准备都很充足，会有合理的限流阀值；如果有突发流量，可能配合监控及时增加机器，横向扩展

总结

限流与熔断，一个让系统保持最大活跃性，一个保障安全，两者技术手段有重叠，但意义完全不同，所以两者都是微服务中不可少的两个服务治理手段

1. semaphore的定义，意义
2. 在没有juc semaphore之前怎么实现
3. semaphore使用
4. 分布式semaphore实现

信号量

最早用来解决进程同步与互斥问题的机制:
包括一个称为信号量的变量及对它进行的两个原语操作(PV操作)

什么是信号量？

信号量（semaphore）的数据结构为一个值和一个指针，指针指向等待该信号量的下一个进程。信号量的值与相应资源的使用情况有关。

PV操作由P操作原语和V操作原语组成（原语是不可中断的过程）

(注,P是荷兰语的Passeren，相当于英文的pass,V是荷兰语的Verhoog,相当于英文中的incremnet)

对信号量进行操作，具体定义如下：

• P（S）：
  • ①将信号量S的值减1，即S=S-1；
  • ②如果S>=0，则该进程继续执行；否则该进程置为等待状态，排入等待队列
• V（S）：
  • ①将信号量S的值加1，即S=S+1；
  • ②如果S>0，则该进程继续执行；否则释放队列中第一个等待信号量的进程

PV操作的意义：我们用信号量及PV操作来实现进程的同步和互斥。PV操作属于进程的低级通信

使用PV操作实现进程互斥时应该注意的是：

1. 每个程序中用户实现互斥的P、V操作必须成对出现，先做P操作，进临界区，后做V操作，出临界区。若有多个分支，要认真检查其成对性
2. P、V操作应分别紧靠临界区的头尾部，临界区的代码应尽可能短，不能有死循环
3. 互斥信号量的初值一般为1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

//许可数量
private int permits = 1;

public synchronized void P() {
    permits--;
    if(permits < 0 ){
        try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public synchronized void V(){
    permits++;
    if(permits <=0){
        notifyAll();
    }
}

J.U.C Semaphore

JUC提供了工具类之一就是Semaphore，提供了丰富的API，不再需要自己实现

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

// 创建具有给定的许可数和非公平的公平设置的 Semaphore。
Semaphore(int permits)
// 创建具有给定的许可数和给定的公平设置的 Semaphore。
Semaphore(int permits, boolean fair)
 
// 从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
void acquire()
// 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞，或者线程已被中断。
void acquire(int permits)
// 从此信号量中获取许可，在有可用的许可前将其阻塞。
void acquireUninterruptibly()
// 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞。
void acquireUninterruptibly(int permits)
// 返回此信号量中当前可用的许可数。
int availablePermits()
// 获取并返回立即可用的所有许可。
int drainPermits()
// 返回一个 collection，包含可能等待获取的线程。
protected Collection<Thread> getQueuedThreads()
// 返回正在等待获取的线程的估计数目。
int getQueueLength()
// 查询是否有线程正在等待获取。
boolean hasQueuedThreads()
// 如果此信号量的公平设置为 true，则返回 true。
boolean isFair()
// 根据指定的缩减量减小可用许可的数目。
protected void reducePermits(int reduction)
// 释放一个许可，将其返回给信号量。
void release()
// 释放给定数目的许可，将其返回到信号量。
void release(int permits)
// 返回标识此信号量的字符串，以及信号量的状态。
String toString()
// 仅在调用时此信号量存在一个可用许可，才从信号量获取许可。
boolean tryAcquire()
// 仅在调用时此信号量中有给定数目的许可时，才从此信号量中获取这些许可。
boolean tryAcquire(int permits)
// 如果在给定的等待时间内此信号量有可用的所有许可，并且当前线程未被中断，则从此信号量获取给定数目的许可。
boolean tryAcquire(int permits, long timeout, TimeUnit unit)
// 如果在给定的等待时间内，此信号量有可用的许可并且当前线程未被中断，则从此信号量获取一个许可。
boolean tryAcquire(long timeout, TimeUnit unit)

对于JUC的Semaphore源码，此篇不阐述了，另开新篇；但对分布式的Semaphore倒是可以研究下

分布式Semaphore

Redission中有对应的RSemaphore

1
2
3
4
5
6

RSemaphore semaphore = redisson.getSemaphore("semaphore");
semaphore.acquire();
//或
semaphore.acquireAsync();
semaphore.acquire(23);
semaphore.tryAcquire();

可过期信号量

1
2
3
4
5
6

RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore");
String permitId = semaphore.acquire();
// 获取一个信号，有效期只有2秒钟。
String permitId = semaphore.acquire(2, TimeUnit.SECONDS);
// ...
semaphore.release(permitId);

直接上最本质的源码片段，lua脚本很简单，对信号量进行计数，acquire时，信号量减1，release时，信号量加1；主要是保证操作的原子性

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

@Override
public RFuture<Boolean> tryAcquireAsync(int permits) {
    if (permits < 0) {
        throw new IllegalArgumentException("Permits amount can't be negative");
    }
    if (permits == 0) {
        return RedissonPromise.newSucceededFuture(true);
    }

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
              "local value = redis.call('get', KEYS[1]); " +
              "if (value ~= false and tonumber(value) >= tonumber(ARGV[1])) then " +
                  "local val = redis.call('decrby', KEYS[1], ARGV[1]); " +
                  "return 1; " +
              "end; " +
              "return 0;",
              Collections.<Object>singletonList(getName()), permits);
}

@Override
public RFuture<Void> releaseAsync(int permits) {
    if (permits < 0) {
        throw new IllegalArgumentException("Permits amount can't be negative");
    }
    if (permits == 0) {
        return RedissonPromise.newSucceededFuture(null);
    }

    return commandExecutor.evalWriteAsync(getName(), StringCodec.INSTANCE, RedisCommands.EVAL_VOID,
        "local value = redis.call('incrby', KEYS[1], ARGV[1]); " +
        "redis.call('publish', KEYS[2], value); ",
        Arrays.<Object>asList(getName(), getChannelName()), permits);
}

在最本质的基础上，再深入看一下还做了哪些事，能真正达到一个工业生产标准

tryAcquire()

非阻塞式，有信息量就正常获取，没有刚快速返回，就是lua本质，没有做额外的事情

acquire()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

@Override
public void acquire(int permits) throws InterruptedException {
    if (tryAcquire(permits)) {
        return;
    }
    RFuture<RedissonLockEntry> future = subscribe();
    commandExecutor.syncSubscription(future);
    try {
        while (true) {
            if (tryAcquire(permits)) {
                return;
            }
            getEntry().getLatch().acquire(permits);
        }
    } finally {
        unsubscribe(future);
    }
}

阻塞式，相对非阻塞式就多了一些事

• 1.先tryAcquire，看是否能获取到信号量
• 2.订阅channel事件
• 3.无限循环
  • 3.1.先tryAcquire()，尝试一下
  • 3.2.通过getEntry().getLatch()，也就是j.u.c.Semaphore,acquire()阻塞
• 4.取消订阅

订阅事件内部细节，另开篇再说了，他的目的其实就是释放Semaphore

想像一下，同一个client的两个线程A，B 同时需要获取信号量，如果A成功获取，那么B将被Semaphore阻塞住了，何时退出阻塞呢？

就在线程A进行release()之后，会publish,细节可查看上面的release()中的lua脚本，当B监听到事件时，就会调用Semaphore.release(),再次进行tryAcquire()

tryAcquire(int permits, long waitTime, TimeUnit unit)

如果在给定的等待时间内此信号量有可用的所有许可，并且当前线程未被中断，则从此信号量获取给定数目的许可

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

@Override
public boolean tryAcquire(int permits, long waitTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);
    long current = System.currentTimeMillis();

    if (tryAcquire(permits)) {
        return true;
    }

    time -= (System.currentTimeMillis() - current);
    if (time <= 0) {
        return false;
    }

    current = System.currentTimeMillis();
    RFuture<RedissonLockEntry> future = subscribe();
    if (!await(future, time, TimeUnit.MILLISECONDS)) {
        return false;
    }

    try {
        time -= (System.currentTimeMillis() - current);
        if (time <= 0) {
            return false;
        }
        
        while (true) {
            current = System.currentTimeMillis();
            if (tryAcquire(permits)) {
                return true;
            }

            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                return false;
            }

            // waiting for message
            current = System.currentTimeMillis();

            getEntry().getLatch().tryAcquire(permits, time, TimeUnit.MILLISECONDS);

            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                return false;
            }
        }
    } finally {
        unsubscribe(future);
    }
//        return get(tryAcquireAsync(permits, waitTime, unit));
}

其实await(future, time, TimeUnit.MILLISECONDS)是使用的CountDownLatch

如果计数到达零，则返回 true；如果在计数到达零之前超过了等待时间，则返回 false

当前是第一个请求，或者别的释放，那就再往下进入循环

CountDownLatch.await()+Semaphore.tryAcquire()配合使用

每一次等待时间后，都需要检查是否超过等待时间

为什么需要引入CountDownLatch.await()呢？ 都使用Semaphore.tryAcquire()不行吗？这个需要再次深入挖掘了

总结

分布式信号量，原理很明了，主要还是通过lua保障redis操作的原子性

阅读redisson源码，发现里面的操作基本都是异步化，底层又是基于netty，大量使用了future模式，如果不知道future模式，会很绕，debug都会晕掉，所以在深入redisson之前，需要再对future模式温习一下

我们不生产代码,我们是代码的搬运工

前不久，阿里大牛虾总再次抛出了分布式锁的讨论，对照之前项目中实现的redis分布式锁总结一下

天才是1%的灵感，加上99%的汗水；编程是1%的编码，加上99%的在Google/StackOverflow/Github上找代码
残酷的现实是，找来的代码可能深藏bug，而不知

锁

在多核多线程环境中，通过锁机制，在某一个时间点上，只能有一个线程进入临界区代码，从而保证临界区中操作数据的一致性

怎么样才是把好锁？

可以保证在分布式部署的应用集群中，同一个方法在同一时间只能被一台机器上的一个线程执行。
这把锁要是一把可重入锁（避免死锁）
支持阻塞和非阻塞：和 ReentrantLock 一样支持 lock 和 trylock 以及 tryLock(long timeOut)
这把锁最好是一把公平锁（根据业务需求考虑要不要这条）
有高可用的获取锁和释放锁功能
获取锁和释放锁的性能要好

分布式锁三要素

1. 外部存储

   分布式锁是在分布式部署环境中给多个主机提供锁服务，需要另外的存储载体

2. 全局唯一标识

   在多线程环境中，锁可以使一个对象引用，也可以是变量，都有唯一的标识来区分锁保护的不同资源；
   在分布式环境下，也需要，比如对某一特定用户资源操作，业务+userId即可唯一标识

3. 至少有两种状态，获取和释放

   锁至少需要两种状态：加锁（lock）和解锁（unlock）。
   用状态区分当前尝试获取的锁是否已经被其他操作占用，
   被占用只有等待锁释放后才能尝试获取锁并加锁，保护共享资源

实现

理论知识知道得再多，还得落地才行；只要遵从三要素，就能打造一把好锁，不要拘泥于某一种工具。

网上有很多实现方式，主要是”外部存储“使用了不同的组件，比如数据库，redis,zk，由于这些组件各自特性的不同，实现复杂度各有不同

这儿主要说下在实际工作中使用到的两种方式，数据库与redis

数据库

数据库，任何系统都需要的组件，常规手法，都是使用version来实现乐观锁

version

比如A、B操作员同时读取一余额为1000元的账户，A操作员为该账户增加100元，B操作员同时为该账户扣除50元，A先提交，B后提交。最后实际账户余额为1000-50=950元，但本该为1000+100-50=1050。这就是典型的并发问题

假设数据库中帐户信息表中有一个version字段，当前值为1；而当前帐户余额字段(balance)为1000元。假设操作员A先更新完，操作员B后更新。
a、操作员A此时将其读出(version=1)，并从其帐户余额中增加100(1000+100=1100)。
b、在操作员A操作的过程中，操作员B也读入此用户信息(version=1)，并从其帐户余额中扣除50(1000-50=950)。
c、操作员A完成了修改工作，将数据版本号加一(version=2)，连同帐户增加后余额(balance=1100)，提交至数据库更新，此时由于提交数据版本大于数据库记录当前版本，数据被更新，数据库记录version更新为2。
d、操作员B完成了操作，也将版本号加一(version=2)试图向数据库提交数据(balance=950)，但此时比对数据库记录版本时发现，操作员B提交的数据版本号为2，数据库记录当前版本也为2，不满足 “提交版本必须大于记录当前版本才能执行更新 “的乐观锁策略，因此，操作员B的提交被驳回。
这样，就避免了操作员B用基于version=1的旧数据修改的结果覆盖操作员A的操作结果的可能。

1

set balance=1100,version=version+1 where id=#{id} and version=#{version};

version简单，除了对业务数据表有侵入性，还有一些场景是胜任不了

比如，在操作一个数量之前，需要确认一下能不能操作

1
2
3
4
5
6
7

int countLimit = select count from limit where id = ${id};

if(countlimit>0){
    set balance=1100,version=version+1 where id=#{id} and version=#{version};
}

update count;

这儿操作了多张表，此时就需要再配合事务，才能保证原子性

redis

由于db性能的限制，而redis性能卓越，很多时候会选择redis实现方式

怎么使用redis正确地实现分布式锁，需要了解两方面

1. 实现分布式锁时，使用到的redis命令
2. 网上示例可能都有毒

redis命令

setnx 命令（『SET if Not eXists』(如果不存在，则 SET)的简写）：
设置成功，返回 1
设置失败，返回 0
该命令是原子操作

getset 命令：
自动将key对应到value并且返回原来key对应的value。如果key存在但是对应的value不是字符串，就返回错误。
返回值：返回之前的旧值，如果之前Key不存在将返回nil。
该命令是原子操作。

get 命令：
get 获取key的值，如果存在，则返回；如果不存在，则返回nil；

del 命令：
del 删除key及key对应的值，如果key不存在，程序忽略

SET 命令：
set key value [EX seconds] [PX milliseconds] [NX|XX]
将字符串值 value 关联到 key 
如果 key 已经持有其他值， SET 就覆写旧值，无视类型。
对于某个原本带有生存时间（TTL）的键来说， 当 SET 命令成功在这个键上执行时， 这个键原有的 TTL 将被清除。

可选参数从 Redis 2.6.12 版本开始，SET 命令的行为可以通过一系列参数来修改：

EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 

PX millisecond：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 

NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 

XX：只在键已经存在时，才对键进行设置操作。

示例

原来项目中使用分布式锁，整个逻辑：

1. setnx(lockkey, 当前时间+过期超时时间)，如果返回 1，则获取锁成功；如果返回 0 则没有获取到锁，转向 2。
2. get(lockkey) 获取值 oldExpireTime ，并将这个 value 值与当前的系统时间进行比较，如果小于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向 3。
3. 计算 newExpireTime = 当前时间+过期超时时间，然后 getset(lockkey, newExpireTime) 会返回当前 lockkey 的值currentExpireTime。判断 currentExpireTime 与 oldExpireTime 是否相等，如果相等，说明当前 getset 设置成功，获取到了锁。如果不相等，说明这个锁又被别的请求获取走了，那么当前请求可以直接返回失败，或者继续重试。
4. 在获取到锁之后，当前线程可以开始自己的业务处理，当处理完毕后，比较自己的处理时间和对于锁设置的超时时间，如果小于锁设置的超时时间，则直接执行 delete 释放锁；如果大于锁设置的超时时间，则不需要再锁进行处理。

获取锁

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

private boolean acquireLock(Jedis j,String lock) throws Exception{
    int timeOut = timeoutSeconds*1000;
    boolean acquired = false;
    long start = System.currentTimeMillis();
    int times = 0;
    do {
        String value = String.valueOf(System.currentTimeMillis() + timeOut + 1);
        // 第一个得到这个锁
        if (j.setnx(lock, value) == 1) {
            logger.info("第一次获取全局锁:{} 成功", lock);
            acquired = true;
            break;
        }
        // j.expire(lock, timeoutSeconds); 网络抖动，可能失败
        String currentValue = j.get(lock);

        // 小于时，可能是上次没有清除，自上次超时后没有别的线程操作过
        if (currentValue != null && Long.valueOf(currentValue) < System.currentTimeMillis()) {
            // 这是同步操作，只会一个成功
            String oldValue = j.getSet(lock, value);
            // 别的线程没有赋上值，当前成功得到锁
            if (oldValue != null && oldValue.equals(currentValue)) {
                acquired = true;
                logger.info("获取全局锁:{} 成功，尝试了{}次,经过了{}ms",lock,times,System.currentTimeMillis()-start);
                break;
            }
        }
        times++;
        Thread.sleep(100);
    } while (start + timeOut > System.currentTimeMillis());
    if(!acquired){
        logger.info("获取全局锁:{} 失败，尝试了{}次",lock,times);
    }			
    return acquired;
}

解锁

1
2
3
4
5
6
7
8
9

private void releaseLock(Jedis j,String lock){
    String currentValue = j.get(lock);
    if(currentValue != null){
        if(System.currentTimeMillis() < Long.valueOf(currentValue) ){
            j.del(lock);
            logger.info("释放锁{}",lock);
        }
    }
}

示例缺陷

特地从多年前的项目中把这段代码找出来，当年写完，心里还挺美

网上有很多资料也是差不多样的，但事实并不那么完美，甚至是错误的

加锁

• 使用jedis.setnx()和jedis.expire()组合实现加锁

1
2
3
4
5

Long result = jedis.setnx(lockKey, value); 
if (result == 1) { 
    // 若在这里程序突然崩溃，则无法设置过期时间，将发生死锁 
    jedis.expire(lockKey, expireTime); 
}

这个问题很明显，setnx与expire不是同一个事务，不俱备原子性；程序崩溃或者网络抖动都会出现死锁问题

• System.currentTimeMillis()
  这个需要各个client时间必须一致，一旦不一致，就可能加锁失败

• getSet()
  如果锁为了灵活性，会把timeout作为入参

当锁过期的时候，如果多个客户端同时执行jedis.getSet()方法，那么虽然最终只有一个客户端可以加锁，但是这个客户端的锁的过期时间可能被其他客户端覆盖

解锁

• jedis.del()直接删除

这种不先判断锁的拥有者而直接解锁的方式，会导致任何客户端都可以随时进行解锁，即使这把锁不是它的

有种错误改进,增加参数传入requestId

1
2
3
4
5
6
7

public static void releaseLock(Jedis jedis, String lockKey, String requestId) { 
    // 判断加锁与解锁是不是同一个客户端 
    if (requestId.equals(jedis.get(lockKey))) { 
        // 若在此时，这把锁突然不是这个客户端的，则会误解锁 
        jedis.del(lockKey);
    } 
}

还是原子性的问题
如代码注释，问题在于如果调用jedis.del()方法的时候，这把锁已经不属于当前客户端的时候会解除他人加的锁。那么是否真的有这种场景？答案是肯定的，比如客户端A加锁，一段时间之后客户端A解锁，在执行jedis.del()之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del()方法，则将客户端B的锁给解除了

缺陷总结

心里认为本来很简单的事，代码大概：

1
2
3
4
5
6
7
8

Lock lock = DistributedReentrantLock.newLock("testlock11");//定义testlock11为key的锁，默认可重入锁
if(lock.tryLock()){
    try{
     xxxxxx
    }finally{
      lock.unlock(); //释放testlock11为key的锁，释放需要放在finally里，防止出异常导致锁没有及时释放
    }
  }

为了提高性能，通过redis原子性接口SETNX:

1. 使用SETNX命令获取锁，若返回0（key已存在，锁已存在）则获取失败，反之获取成功
2. 为了防止获取锁后程序出现异常，导致其他线程/进程调用SETNX命令总是返回0而进入死锁状态，需要为该key设置一个“合理”的过期时间释放锁
3. 使用DEL命令将锁数据删除

结果为了弥补setnx()与expire()两个接口的原子性问题，引入了一堆问题，外强中干

缺陷修正

加锁

Redis 2.6.12版本后，增强了set()命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

/**
 * 尝试获取分布式锁
 * @param jedis Redis客户端
 * @param lockKey 锁
 * @param requestId 请求标识
 * @param expireTime 超期时间
 * @return 是否获取成功
 */
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
	String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
	if (LOCK_SUCCESS.equals(result)) {
		return true;
	}
	return false;
}

加锁就一行代码：jedis.set(String key, String value, String nxxx, String expx, int time)，
这个set()方法一共有五个入参：

1. 第一个为key，我们使用key来当锁，因为key是唯一的
2. 第二个为value，我们传的是requestId，通过给value赋值为requestId，就知道这把锁是哪个请求加的了，在解锁的时候就可以有依据。requestId可以使用UUID.randomUUID().toString()方法生成
3. 第三个为nxxx，这个参数我们填的是NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作；
4. 第四个为expx，这个参数我们传的是PX，意思是我们要给这个key加一个过期的设置，具体时间由第五个参数决定
5. 第五个为time，与第四个参数相呼应，代表key的过期时间

高可用：

1. set()加入了NX参数，可以保证如果已有key存在，则不会调用成功，也就是只有一个客户端能持有锁，满足互斥性
2. 由于我们对锁设置了过期时间，即使锁的持有者后续发生崩溃而没有解锁，锁也会因为到了过期时间而自动解锁（即key被删除），不会发生死锁
3. 将value赋值为requestId，代表加锁的客户端请求标识，那么在解锁的时候就可以进行校验是否是同一个客户端，防止锁交叉

解锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

/**
 * 释放分布式锁
 * @param jedis Redis客户端
 * @param lockKey 锁
 * @param requestId 请求标识
 * @return 是否释放成功
 */
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {

	String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
	Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));

	if (RELEASE_SUCCESS.equals(result)) {
		return true;
	}
	return false;
}

首先获取锁对应的value值，检查是否与requestId相等，如果相等则删除锁（解锁）

使用eval()配置lua保证原子性

在eval命令执行Lua代码的时候，Lua代码将被当成一个命令去执行，并且直到eval命令执行完成，Redis才会执行其他命令

有效时间

为什么需要一个有效时间呢？主要就是防止死锁

疑难

• 执行业务代码操作共享资源的时间大于设置锁的过期时间？

客户端需要设置接口访问超时,接口超时时间需要远远小于锁超时时间,比如锁自动释放的时间是10s,那么接口超时大概设置5-50ms

【虽然能解决问题，但时间设置成了难点，微服务中多少接口，而且接口的timeout都是可配置的，不能每次调整接口timeout时，还是考虑一下锁的timeout】

• GC的STW

客户端1获得了锁，正准备处理共享资源的时候，发生了Full GC直到锁过期。这样，客户端2又获得了锁，开始处理共享资源。在客户端2处理的时候，客户端1 Full GC完成，也开始处理共享资源，这样就出现了2个客户端都在处理共享资源的情况

续命丸

引入锁续约机制，也就是获取锁之后，释放锁之前，会定时进行锁续约，比如以3min间隔周期进行锁续约

这样如果应用重启了，最多3min等待时间，不会因为时间太长导致的死锁问题，也不会因为时间太短导致被其他线程抢占的问题，也就是锁分布式锁不需要设置过期时间，过期时间对于这个锁来说是滑动的

Redission

虾总给了总结性阐述：

首先启动Daemon线程，一直循环检测所有的分布式key，异步递延分布锁的过期时间，只要在处理业务逻辑，就递延分布锁过期时间3min。
每次添加分布式锁key，同时会生成一个uuid token，定义一个ConcurrentHashMap构造一个全局map维护所有的分布式key，上面Daemon线程会遍历这个map，每次解锁需要比对这个token，token一致才能解锁。
这样以来如果应用重启了，最多会有3min等待时间，不会导致时间太长导致的死锁问题，也不会因为时间太短导致的被其他线程抢占的问题，也就是锁分布式锁不需要设置过期时间，过期时间对于这个锁来说是滑动的

跟随虾总思路，找到了一个开源组件：Redisson

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。

相对于平时使用的jedis，redission进行比较高的抽象

redission中的lock主要是RLock接口，继承的juc的Lock接口

1

public interface RLock extends Lock, RExpirable, RLockAsync 

Lock

先看lock(),有两种形式，一个不带leaseTime,一个带leaseTime

1
2

public void lock() ;
public void lock(long leaseTime, TimeUnit unit) ;

边看源码，边解释

两个方法共用了lockInterruptibly()

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

public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
	long threadId = Thread.currentThread().getId();
	Long ttl = tryAcquire(leaseTime, unit, threadId);
	// lock acquired
	if (ttl == null) {
		return;
	}
	RFuture<RedissonLockEntry> future = subscribe(threadId);
	commandExecutor.syncSubscription(future);
	try {
		while (true) {
			ttl = tryAcquire(leaseTime, unit, threadId);
			// lock acquired
			if (ttl == null) {
				break;
			}
			// waiting for message
			if (ttl >= 0) {
				getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
			} else {
				getEntry(threadId).getLatch().acquire();
			}
		}
	} finally {
		unsubscribe(future, threadId);
	}
//        get(lockAsync(leaseTime, unit));
}

1. 尝试获取锁tryAcquire
2. 获取失败，订阅此channel的消息（订阅的意义，在解锁时就会发现）
3. 进入循环，不停的尝试获取锁，其中使用了JUC的Semaphore
4. 一旦获取成功，则跳出循环
5. 取消订阅

尝试获取锁tryAcquire里面会用到两个核心方法tryAcquireAsync(),tryLockInnerAsync()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

@Override
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
	if (leaseTime != -1) {
		return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
	}
	RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
	ttlRemainingFuture.addListener(new FutureListener<Long>() {
		@Override
		public void operationComplete(Future<Long> future) throws Exception {
			if (!future.isSuccess()) {
				return;
			}

			Long ttlRemaining = future.getNow();
			// lock acquired
			if (ttlRemaining == null) {
				scheduleExpirationRenewal(threadId);
			}
		}
	});
	return ttlRemainingFuture;
}

• 1.根据锁的持续时间不同，处理也不同
• 2.没有设置持续时间，那就是阻塞型，一直等待
  • 2.1.为了防止业务方法执行时间超过锁timeout,则定时续约scheduleExpirationRenewal()
• 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

private void scheduleExpirationRenewal(final long threadId) {
	if (expirationRenewalMap.containsKey(getEntryName())) {
		return;
	}
	Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
		@Override
		public void run(Timeout timeout) throws Exception {
			
			RFuture<Boolean> future = renewExpirationAsync(threadId);
			
			future.addListener(new FutureListener<Boolean>() {
				@Override
				public void operationComplete(Future<Boolean> future) throws Exception {
					expirationRenewalMap.remove(getEntryName());
					if (!future.isSuccess()) {
						log.error("Can't update lock " + getName() + " expiration", future.cause());
						return;
					}
					
					if (future.getNow()) {
						// reschedule itself
						scheduleExpirationRenewal(threadId);
					}
				}
			});
		}

	}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

	if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
		task.cancel();
	}
}

protected RFuture<Boolean> renewExpirationAsync(long threadId) {
	return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
			"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
				"redis.call('pexpire', KEYS[1], ARGV[1]); " +
				"return 1; " +
			"end; " +
			"return 0;",
		Collections.<Object>singletonList(getName()), 
		internalLockLeaseTime, getLockName(threadId));
}

1. 以internalLockLeaseTime/3间隔时间，定时续约
2. 如果当前client自身有并发时，通过putIfAbsent保证只有一个task
3. 续约：当lock存在时，使用pexpire设置过期时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
	internalLockLeaseTime = unit.toMillis(leaseTime);

	return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
			  "if (redis.call('exists', KEYS[1]) == 0) then " +
				  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
				  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
				  "return nil; " +
			  "end; " +
			  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
				  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
				  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
				  "return nil; " +
			  "end; " +
			  "return redis.call('pttl', KEYS[1]);",
				Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

protected String getLockName(long threadId) { 
    return id + ":" + threadId;
}

• 1.lockname不存在
  • 1.1.hset(lockname,uuid+threadid,1),value=uuid+threadid，有uuid可以区分各个client，有threadid区分各个线程，这样锁就具备了可重入性
  • 1.2.pexpire设置过期时间，防止client挂掉，造成死锁
• 2.lockname存在
  • 2.1.hexists(lockname,uuid+threadid),这样保证了是同一个锁在同一个client
  • 2.2.hincrby 再次进锁，计数器+1
  • 2.3.pexpire 再次设置超时
• 3.lockname存在，并且不在同一client
  • 3.1.pttl 返回剩余有效时长

unLock

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

@Override
public RFuture<Void> unlockAsync(final long threadId) {
	final RPromise<Void> result = new RedissonPromise<Void>();
	RFuture<Boolean> future = unlockInnerAsync(threadId);

	future.addListener(new FutureListener<Boolean>() {
		@Override
		public void operationComplete(Future<Boolean> future) throws Exception {
			if (!future.isSuccess()) {
				cancelExpirationRenewal(threadId);
				result.tryFailure(future.cause());
				return;
			}

			Boolean opStatus = future.getNow();
			if (opStatus == null) {
				IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
						+ id + " thread-id: " + threadId);
				result.tryFailure(cause);
				return;
			}
			if (opStatus) {
				cancelExpirationRenewal(null);
			}
			result.trySuccess(null);
		}
	});

	return result;
}

void cancelExpirationRenewal(Long threadId) {
	ExpirationEntry task = expirationRenewalMap.get(getEntryName());
	if (task != null && (threadId == null || task.getThreadId() == threadId)) {
		expirationRenewalMap.remove(getEntryName());
		task.getTimeout().cancel();
	}
}

1. 从方法名看，虽然对外好像是直接解锁，但内部是异步执行的
2. unlockInnerAsync()进行解锁
3. 从expirationRenewalMap移除，并把task.cancel()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
	return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
			"if (redis.call('exists', KEYS[1]) == 0) then " +
				"redis.call('publish', KEYS[2], ARGV[1]); " +
				"return 1; " +
			"end;" +
			"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
				"return nil;" +
			"end; " +
			"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
			"if (counter > 0) then " +
				"redis.call('pexpire', KEYS[1], ARGV[2]); " +
				"return 0; " +
			"else " +
				"redis.call('del', KEYS[1]); " +
				"redis.call('publish', KEYS[2], ARGV[1]); " +
				"return 1; "+
			"end; " +
			"return nil;",
			Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

}

1. lockname不存在，说明已经解锁，publish channelname unlockmessage；return 1
2. lockname存在，但对于uuid+id不存在，说明不是加锁的client,return nil
3. lockname存在，并且是当前加锁client
4. 对lockname uuid+id进行-1，如果counter>0则走5，如果=0 则走6
5. counter>0 说明锁重入了，计数器-1，并expire
6. counter=0 说明最终解锁，直接del key，并publish channelname unlockmessage;return 1

redission缺陷

使用cluster时

一个场景：A在向主机1请求到锁成功后，主机1宕机了。现在从机1a变成了主机。但是数据没有同步，从机1a是没有A的锁的。那么B又可以获得一个锁。这样就会造成数据错误。

redlock主要思想就是做数据冗余。建立5台独立的集群，当我们发送一个数据的时候，要保证3台（n/2+1）以上的机器接受成功才算成功，否则重试或报错

redlock实现会更复杂，但从他的算法上看，有zk选举的味道。对于更高可用分布锁，可以借助zk本身特性去实现

总结

对于锁，主要考虑性能与安全，即要保持锁的活跃性，又得保证锁的安全性

分布式锁，除了以上两点，还要考虑实现时的三要素

对于redission，对于锁部分的源码，还有很多的内容，很多的细节需要挖掘，此篇就不写了，太长。

后面再结合JUC，写篇更详细的源码分析

参考资料

Redis分布式锁的正确实现方式

redission