转载:深入浅出 RPC - 深入篇

RPC 是什么？

RPC 的全称是 Remote Procedure Call 是一种进程间通信方式。它允许程序调用另一个地址空间（通常是共享网络的另一台机器上）的过程或函数，而不用程序员显式编码这个远程调用的细节。即程序员无论是调用本地的还是远程的，本质上编写的调用代码基本相同。

RPC 起源

RPC 这个概念术语在上世纪 80 年代由 Bruce Jay Nelson 提出。在 Nelson 的论文 “Implementing Remote Procedure Calls” 中他提到了几点：

• 简单：RPC 概念的语义十分清晰和简单，这样建立分布式计算就更容易。
• 高效：过程调用看起来十分简单而且高效。
• 通用：在单机计算中过程往往是不同算法部分间最重要的通信机制。

通俗一点说，就是一般程序员对于本地的过程调用很熟悉，那么我们把 RPC 作成和本地调用完全类似，那么就更容易被接受，使用起来毫无障碍。今天我们使用的 RPC 框架基本就是按这个目标来实现的。

RPC 结构

Nelson 的论文中指出实现 RPC 的程序包括 5 个部分：

1. User
2. User-stub
3. RPCRuntime
4. Server-stub
5. Server

这 5 个部分的关系如下图所示:

这里 user 就是 client 端，当 user 想发起一个远程调用时，它实际是通过本地调用 user-stub。user-stub 负责将调用的接口、方法和参数通过约定的协议规范进行编码并通过本地的 RPCRuntime 实例传输到远端的实例。远端 RPCRuntime 实例收到请求后交给 server-stub 进行解码后发起本地端调用，调用结果再返回给 user 端。

RPC 功能目标

RPC 的主要功能目标是让构建分布式计算（应用）更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。为实现该目标，RPC 框架需提供一种透明调用机制让使用者不必显式的区分本地调用和远程调用。

RPC 调用分类

RPC调用分以下两种：

1. 同步调用

   客户方等待调用执行完成并返回结果。

2. 异步调用

   客户方调用后不用等待执行结果返回，但依然可以通过回调通知等方式获取返回结果。

   若客户方不关心调用返回结果，则变成单向异步调用，单向调用不用返回结果。

异步和同步的区分在于是否等待服务端执行完成并返回结果。

RPC 结构拆解

RPC服务方通过 RpcServer 去导出（export）远程接口方法，而客户方通过 RpcClient 去引入（import）远程接口方法。客户方像调用本地方法一样去调用远程接口方法，RPC 框架提供接口的代理实现，实际的调用将委托给代理RpcProxy 。代理封装调用信息并将调用转交给RpcInvoker 去实际执行。在客户端的RpcInvoker 通过连接器RpcConnector 去维持与服务端的通道RpcChannel，并使用RpcProtocol 执行协议编码（encode）并将编码后的请求消息通过通道发送给服务方。

RPC 服务端接收器 RpcAcceptor 接收客户端的调用请求，同样使用RpcProtocol 执行协议解码（decode）。解码后的调用信息传递给RpcProcessor 去控制处理调用过程，最后再委托调用给RpcInvoker 去实际执行并返回调用结果。

RPC 组件职责

下面详细说明下每个组件的职责划分。

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

1. RpcServer  
   负责导出（export）远程接口  
2. RpcClient  
   负责导入（import）远程接口的代理实现  
3. RpcProxy  
   远程接口的代理实现  
4. RpcInvoker  
   客户方实现：负责编码调用信息和发送调用请求到服务方并等待调用结果返回  
   服务方实现：负责调用服务端接口的具体实现并返回调用结果  
5. RpcProtocol  
   负责协议编/解码  
6. RpcConnector  
   负责维持客户方和服务方的连接通道和发送数据到服务方  
7. RpcAcceptor  
   负责接收客户方请求并返回请求结果  
8. RpcProcessor  
   负责在服务方控制调用过程，包括管理调用线程池、超时时间等  
9. RpcChannel  
   数据传输通道

RPC 实现分析

导出远程接口

导出远程接口的意思是指只有导出的接口可以供远程调用，而未导出的接口则不能。在 java 中导出接口的代码片段可能如下：

1
2
3

DemoService demo   = new DemoService();  
RpcServer  server = new RpcServer();  
server.export(DemoService.class, demo, options);

我们可以导出整个接口，也可以更细粒度一点只导出接口中的某些方法，如：

1
2

// 只导出 DemoService 中签名为 hi(String s) 的方法  
server.export(DemoService.class, demo, "hi", new Class<?>[] { String.class }, options);

java 中还有一种比较特殊的调用就是多态，也就是一个接口可能有多个实现，那么远程调用时到底调用哪个？这个本地调用的语义是通过 jvm 提供的引用多态性隐式实现的，那么对于 RPC 来说跨进程的调用就没法隐式实现了。如果前面DemoService 接口有 2 个实现，那么在导出接口时就需要特殊标记不同的实现，如：

1
2
3
4
5

DemoService demo   = new DemoService();  
DemoService demo2  = new DemoService();  
RpcServer   server = new RpcServer();  
server.export(DemoService.class, demo, options);  
server.export("demo2", DemoService.class, demo2, options);

上面 demo2 是另一个实现，我们标记为 “demo2” 来导出，那么远程调用时也需要传递该标记才能调用到正确的实现类，这样就解决了多态调用的语义。

导入远程接口与客户端代理

导入相对于导出远程接口，客户端代码为了能够发起调用必须要获得远程接口的方法或过程定义。目前，大部分跨语言平台 RPC 框架采用根据 IDL 定义通过 code generator 去生成 stub 代码，这种方式下实际导入的过程就是通过代码生成器在编译期完成的。

代码生成的方式对跨语言平台 RPC 框架而言是必然的选择，而对于同一语言平台的 RPC 则可以通过共享接口定义来实现。在 java 中导入接口的代码片段可能如下：

1
2
3

RpcClient client = new RpcClient();  
DemoService demo = client.refer(DemoService.class);  
demo.hi("how are you?");

在 java 中 ‘import’ 是关键字，所以代码片段中我们用 refer 来表达导入接口的意思。这里的导入方式本质也是一种代码生成技术，只不过是在运行时生成，比静态编译期的代码生成看起来更简洁些。

协议编解码

客户端代理在发起调用前需要对调用信息进行编码，这就要考虑需要编码些什么信息并以什么格式传输到服务端才能让服务端完成调用。出于效率考虑，编码的信息越少越好（传输数据少），编码的规则越简单越好（执行效率高）。我们先看下需要编码些什么信息：

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

-- 调用编码 --  
1. 接口方法  
   包括接口名、方法名  
2. 方法参数  
   包括参数类型、参数值  
3. 调用属性  
   包括调用属性信息，例如调用附件隐式参数、调用超时时间等  
  
-- 返回编码 --  
1. 返回结果  
   接口方法中定义的返回值  
2. 返回码  
   异常返回码  
3. 返回异常信息  
   调用异常信息

除了以上这些必须的调用信息，我们可能还需要一些元信息以方便程序编解码以及未来可能的扩展。这样我们的编码消息里面就分成了两部分，一部分是元信息、另一部分是调用的必要信息。如果设计一种 RPC 协议消息的话，元信息我们把它放在协议消息头中，而必要信息放在协议消息体中。下面给出一种概念上的 RPC 协议消息设计格式：

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

-- 消息头 --  
magic      : 协议魔数，为解码设计  
header size: 协议头长度，为扩展设计  
version    : 协议版本，为兼容设计  
st         : 消息体序列化类型  
hb         : 心跳消息标记，为长连接传输层心跳设计  
ow         : 单向消息标记，  
rp         : 响应消息标记，不置位默认是请求消息  
status code: 响应消息状态码  
reserved   : 为字节对齐保留  
message id : 消息 id  
body size  : 消息体长度  
  
-- 消息体 --  
采用序列化编码，常见有以下格式  
xml   : 如 webservie soap  
json  : 如 JSON-RPC  
binary: 如 thrift; hession; kryo 等

格式确定后编解码就简单了，由于头长度一定所以我们比较关心的就是消息体的序列化方式。序列化我们关心三个方面：

1. 序列化和反序列化的效率，越快越好。
2. 序列化后的字节长度，越小越好。
3. 序列化和反序列化的兼容性，接口参数对象若增加了字段，是否兼容。

传输服务

协议编码之后，自然就是需要将编码后的 RPC 请求消息传输到服务方，服务方执行后返回结果消息或确认消息给客户方。RPC 的应用场景实质是一种可靠的请求应答消息流，和 HTTP 类似。因此选择长连接方式的 TCP 协议会更高效，与 HTTP 不同的是在协议层面我们定义了每个消息的唯一 id，因此可以更容易的复用连接。

既然使用长连接，那么第一个问题是到底 client 和 server 之间需要多少根连接？实际上单连接和多连接在使用上没有区别，对于数据传输量较小的应用类型，单连接基本足够。单连接和多连接最大的区别在于，每根连接都有自己私有的发送和接收缓冲区，因此大数据量传输时分散在不同的连接缓冲区会得到更好的吞吐效率。所以，如果你的数据传输量不足以让单连接的缓冲区一直处于饱和状态的话，那么使用多连接并不会产生任何明显的提升，反而会增加连接管理的开销。

连接是由 client 端发起建立并维持。如果 client 和 server 之间是直连的，那么连接一般不会中断（当然物理链路故障除外）。如果 client 和 server 连接经过一些负载中转设备，有可能连接一段时间不活跃时会被这些中间设备中断。为了保持连接有必要定时为每个连接发送心跳数据以维持连接不中断。心跳消息是 RPC 框架库使用的内部消息，在前文协议头结构中也有一个专门的心跳位，就是用来标记心跳消息的，它对业务应用透明。

执行调用

client stub 所做的事情仅仅是编码消息并传输给服务方，而真正调用过程发生在服务方。server stub 从前文的结构拆解中我们细分了 RpcProcessor 和 RpcInvoker 两个组件，一个负责控制调用过程，一个负责真正调用。这里我们还是以 java 中实现这两个组件为例来分析下它们到底需要做什么？

java 中实现代码的动态接口调用目前一般通过反射调用。除了原生的 jdk 自带的反射，一些第三方库也提供了性能更优的反射调用，因此 RpcInvoker 就是封装了反射调用的实现细节。

调用过程的控制需要考虑哪些因素，RpcProcessor 需要提供什么样地调用控制服务呢？

1
2
3
4
5
6

1. 效率提升  
   每个请求应该尽快被执行，因此我们不能每请求来再创建线程去执行，需要提供线程池服务。  
2. 资源隔离  
   当我们导出多个远程接口时，如何避免单一接口调用占据所有线程资源，而引发其他接口执行阻塞。  
3. 超时控制  
   当某个接口执行缓慢，而 client 端已经超时放弃等待后，server 端的线程继续执行此时显得毫无意义。

RPC 异常处理

无论 RPC 怎样努力把远程调用伪装的像本地调用，但它们依然有很大的不同点，而且有一些异常情况是在本地调用时绝对不会碰到的。在说异常处理之前，我们先比较下本地调用和 RPC 调用的一些差异：

1. 本地调用一定会执行，而远程调用则不一定，调用消息可能因为网络原因并未发送到服务方。
2. 本地调用只会抛出接口声明的异常，而远程调用还会跑出 RPC 框架运行时的其他异常。
3. 本地调用和远程调用的性能可能差距很大，这取决于 RPC 固有消耗所占的比重。

正是这些区别决定了使用 RPC 时需要更多考量。当调用远程接口抛出异常时，异常可能是一个业务异常，也可能是 RPC 框架抛出的运行时异常（如：网络中断等）。业务异常表明服务方已经执行了调用，可能因为某些原因导致未能正常执行，而 RPC 运行时异常则有可能服务方根本没有执行，对调用方而言的异常处理策略自然需要区分。

由于 RPC 固有的消耗相对本地调用高出几个数量级，本地调用的固有消耗是纳秒级，而 RPC 的固有消耗是在毫秒级。那么对于过于轻量的计算任务就并不合适导出远程接口由独立的进程提供服务，只有花在计算任务上时间远远高于 RPC 的固有消耗才值得导出为远程接口提供服务。

RPC调用流程

1
2
3
4
5
6
7
8
9
10
11

1）客户端（client）调用以本地调用方式调用服务；
2）client stub接收到调用后负责将方法、参数等组装成能够进行网络传输的消息体；
3）client stub找到服务地址，并将消息发送到服务端；
4）server stub收到消息后进行解码；
5）server stub根据解码结果调用本地的服务；
6）服务端执行并将结果返回给server stub；
7）server stub将返回结果打包成消息并发送至消费方；
8）client stub接收到消息，并进行解码；
9）客户端得到最终结果。

Thrift主要特点

1. 基于二进制的高性能的编解码框架
2. 基于NIO的底层通信
3. 相对简单的服务调用模型
4. 使用IDL支持跨平台调用

Thrift核心组件

1. TProtocol 协议和编解码组件
2. TTransport 传输组件
3. TProcessor 服务调用组件
4. TServer，Client 服务器和客户端组件
5. IDL 服务描述组件，负责生产跨平台客户端

分布式缓存问题

​ 分布式系统中，当服务增长到一定规模时，惯常的做法是集群化，引入负载均衡，这样做的好处是：1. 高可用。2. 解耦。从外部看，透明化了集群的内部细节（外部都通过负载均衡服务器通信，然后由负载均衡服务器分发请求）。

假设一个简单的场景：有4个cache服务器组成的集群，当一个对象传入集群时，这个对象应该存储在哪一个cache里呢？一种简单的方法是使用映射公式：

1

Hash(object) % 4

这个算法的问题在于容错性和扩展性不好。所谓容错性是指当系统中某一个或几个服务器变得不可用时，整个系统是否可以正确高效运行；而扩展性是指当加入新的服务器后，整个系统是否可以正确高效运行。

假设一下情况：

• 由于流量增大，需要增加一台cache，共5个cache。这时，映射公式就变成Hash(object) % 5。
• 有一个cache服务器down掉，变成3个cache。这时，映射公式就变成Hash(object) % 3。

因此系统中一旦有服务器变更，大量的key会被重定位到不同的服务器从而造成大量的缓存不命中，这意味着一时间所有的缓存全部失效。而这种情况在分布式系统中是非常糟糕的。

一个设计良好的分布式哈希方案应该具有良好的单调性，即服务节点的增减不会造成大量哈希重定位。一致性哈希算法就是这样一种哈希方案。

一致性哈希算法

1. 简单来说，一致性哈希将整个哈希值空间组织成一个虚拟的圆环，如假设某哈希函数H的值空间为${2}^{32}$-1（即哈希值是一个32位无符号整形），整个哈希空间环如下：

整个空间按顺时针方向组织。0和${2}^{32}$-1在零点中方向重合。

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

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

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

根据一致性哈希算法，数据A会被定为到Server 1上，D被定为到Server 3上，而B、C分别被定为到Server 2上。

容错性与可扩展性分析

新的一致性hash算法成功解决了cache服务器增减时key的失效问题。现在，无论增减cache，只有部分key失效。

现假设Server 3宕机了：

可以看到此时A、C、B不会受到影响，只有D节点被重定位到Server 2。一般的，在一致性哈希算法中，如果一台服务器不可用，则受影响的数据仅仅是此服务器到其环空间中前一台服务器（即顺着逆时针方向行走遇到的第一台服务器）之间数据，其它不会受到影响。

下面考虑另外一种情况，如果我们在系统中增加一台服务器Memcached Server 4：

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

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

虚拟节点

对于一个object来说，它落在环上的任何位置的概率都是一样的，那么落在一个cache的概率就和圆弧的长度成正比。于是，我们希望每个cache所占的圆弧长度更接近。

理论上，只要cache足够多，每个cache在圆环上就会足够分散。但是在真实场景里，cache服务器只会有很少，所以，引入了“虚拟节点”的概念。

一致性哈希算法在服务节点太少时，容易因为节点分部不均匀而造成数据倾斜问题。例如我们的系统中有两台服务器，其环分布如下：

此时必然造成大量数据集中到Server 1上，而只有极少量会定位到Server 2上。为了解决这种数据倾斜问题，一致性哈希算法引入了虚拟节点机制，将一个物理节点拆分为多个虚拟节点，并且同一个物理节点的虚拟节点尽量均匀分布在Hash环上。具体做法可以在服务器ip或主机名的后面增加编号来实现。

实现一致性哈希(java)

• 一致性Hash算法实现版本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
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

package com.zsr.test.ConsistentHash;
import java.util.Map.Entry;
import java.util.TreeMap;
/**
 * 一致性Hash算法实现版本1：不带虚拟节点
 * 
 * @author David
 */
public class ConsistentHashingWithoutVirtualNode {
  /**
   * 待添加入Hash环的服务器列表
   */
  private static String[] servers = { "192.168.0.0:111", "192.168.0.1:111", "192.168.0.2:111", "192.168.0.3:111",
      "192.168.0.4:111" };
  /**
   * key表示服务器的hash值，value表示服务器的名称
   */
  private static TreeMap<Integer, String> treeMap = new TreeMap<Integer, String>();
  /**
   * 程序初始化，将所有的服务器放入sortedMap中
   */
  static {
    for (int i = 0; i < servers.length; i++) {
      int hash = getHash(servers[i]);
      System.out.println("[" + servers[i] + "]加入集合中, 其Hash值为" + hash);
      treeMap.put(hash, servers[i]);
    }
    System.out.println();
  }
  /**
   * 简单起见，复用String的hashCode算法;实际hash算法可以采用MurmurHash
   */
  private static int getHash(String str) {
    int hash = str.hashCode();
    // String的hashCode()方法却会产生负数，取绝对值
    return Math.abs(hash);
  }
  /**
   * 得到应当路由到的结点
   */
  private static String getServer(String node) {
    // 得到带路由的结点的Hash值
    int hash = getHash(node);
    // 获取大于或等于指定key的最小map键相关联的map键值对
    Entry<Integer, String> entry = treeMap.ceilingEntry(hash);
    if (entry == null) {
      return treeMap.firstEntry().getValue();
    }
    return entry.getValue();
  }
  public static void main(String[] args) {
    String[] nodes = { "127.0.0.1:1111", "221.226.0.1:2222", "10.211.0.1:3333" };
    for (int i = 0; i < nodes.length; i++)
      System.out.println("[" + nodes[i] + "]的hash值为" + getHash(nodes[i]) + ", 被路由到结点[" + getServer(nodes[i]) + "]");
  }
}

测试结果:

1
2
3
4
5
6
7
8
9

[192.168.0.0:111]加入集合中, 其Hash值为771739798
[192.168.0.1:111]加入集合中, 其Hash值为770816277
[192.168.0.2:111]加入集合中, 其Hash值为769892756
[192.168.0.3:111]加入集合中, 其Hash值为768969235
[192.168.0.4:111]加入集合中, 其Hash值为768045714
[127.0.0.1:1111]的hash值为35944419, 被路由到结点[192.168.0.4:111]
[221.226.0.1:2222]的hash值为973393028, 被路由到结点[192.168.0.4:111]
[10.211.0.1:3333]的hash值为561068530, 被路由到结点[192.168.0.4:111]

从运行结果来看，也没有问题，三个点路由到的都是顺时针离他们Hash值最近的那台服务器上。

• 一致性Hash算法实现版本2：带虚拟节点

编程方面需要考虑的问题是：

1. 一个真实结点如何对应成为多个虚拟节点？
2. 虚拟节点找到后如何还原为真实结点？

这两个问题其实有很多解决办法，这里使用了一种简单的办法，给每个真实结点后面根据虚拟节点加上后缀再取Hash值，比如”192.168.0.0:111”就把它变成”192.168.0.0:111&&VN0”到”192.168.0.0:111&&VN4”，VN就是Virtual Node的缩写，还原的时候只需要从头截取字符串到”&&”的位置就可以了。

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

package com.zsr.test.ConsistentHash;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.TreeMap;
/**
 * 带虚拟节点的一致性Hash算法
 * 
 * @author David
 */
public class ConsistentHashingWithVirtualNode {
  /**
   * 待添加入Hash环的服务器列表
   */
  private static String[] servers = { "192.168.0.0:111", "192.168.0.1:111", "192.168.0.2:111", "192.168.0.3:111",
      "192.168.0.4:111" };
  /**
   * 真实结点列表,考虑到服务器上线、下线的场景，即添加、删除的场景会比较频繁，这里使用LinkedList会更好
   */
  private static List<String> realNodes = new LinkedList<String>();
  /**
   * 虚拟节点，key表示虚拟节点的hash值，value表示虚拟节点的名称
   */
  private static TreeMap<Integer, String> virtualNodes = new TreeMap<Integer, String>();
  /**
   * 虚拟节点的数目，这里写死，为了演示需要，一个真实结点对应5个虚拟节点
   */
  private static final int VIRTUAL_NODES = 5;
  static {
    // 先把原始的服务器添加到真实结点列表中
    for (int i = 0; i < servers.length; i++)
      realNodes.add(servers[i]);
    // 再添加虚拟节点，遍历LinkedList使用foreach循环效率会比较高
    for (String str : realNodes) {
      for (int i = 0; i < VIRTUAL_NODES; i++) {
        String virtualNodeName = str + "&&VN" + String.valueOf(i);
        int hash = getHash(virtualNodeName);
        System.out.println("虚拟节点[" + virtualNodeName + "]被添加, hash值为" + hash);
        virtualNodes.put(hash, virtualNodeName);
      }
    }
    System.out.println();
  }
  /**
   * 简单起见，复用String的hashCode算法;实际hash算法可以采用MurmurHash
   */
  private static int getHash(String str) {
    int hash = str.hashCode();
    // String的hashCode()方法却会产生负数，取绝对值
    return Math.abs(hash);
  }
  /**
   * 得到应当路由到的结点
   */
  private static String getServer(String node) {
    // 得到带路由的结点的Hash值
    int hash = getHash(node);
    // 获取大于或等于指定key的最小map键相关联的map键值对
    Entry<Integer, String> entry = virtualNodes.ceilingEntry(hash);
    String virtualNode = null;
    if (entry == null) {
      virtualNode = virtualNodes.firstEntry().getValue();
    }
    virtualNode = entry.getValue();
    return virtualNode.substring(0, virtualNode.indexOf("&&"));
  }
  public static void main(String[] args) {
    String[] nodes = { "127.0.0.1:1111", "221.226.0.1:2222", "10.211.0.1:3333" };
    for (int i = 0; i < nodes.length; i++)
      System.out.println("[" + nodes[i] + "]的hash值为" + getHash(nodes[i]) + ", 被路由到结点[" + getServer(nodes[i]) + "]");
  }
}

测试结果:

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

虚拟节点[192.168.0.0:111&&VN0]被添加, hash值为1490088590
虚拟节点[192.168.0.0:111&&VN1]被添加, hash值为1490088591
虚拟节点[192.168.0.0:111&&VN2]被添加, hash值为1490088592
虚拟节点[192.168.0.0:111&&VN3]被添加, hash值为1490088593
虚拟节点[192.168.0.0:111&&VN4]被添加, hash值为1490088594
虚拟节点[192.168.0.1:111&&VN0]被添加, hash值为1293575085
虚拟节点[192.168.0.1:111&&VN1]被添加, hash值为1293575086
虚拟节点[192.168.0.1:111&&VN2]被添加, hash值为1293575087
虚拟节点[192.168.0.1:111&&VN3]被添加, hash值为1293575088
虚拟节点[192.168.0.1:111&&VN4]被添加, hash值为1293575089
虚拟节点[192.168.0.2:111&&VN0]被添加, hash值为1097061580
虚拟节点[192.168.0.2:111&&VN1]被添加, hash值为1097061581
虚拟节点[192.168.0.2:111&&VN2]被添加, hash值为1097061582
虚拟节点[192.168.0.2:111&&VN3]被添加, hash值为1097061583
虚拟节点[192.168.0.2:111&&VN4]被添加, hash值为1097061584
虚拟节点[192.168.0.3:111&&VN0]被添加, hash值为900548075
虚拟节点[192.168.0.3:111&&VN1]被添加, hash值为900548076
虚拟节点[192.168.0.3:111&&VN2]被添加, hash值为900548077
虚拟节点[192.168.0.3:111&&VN3]被添加, hash值为900548078
虚拟节点[192.168.0.3:111&&VN4]被添加, hash值为900548079
虚拟节点[192.168.0.4:111&&VN0]被添加, hash值为704034570
虚拟节点[192.168.0.4:111&&VN1]被添加, hash值为704034571
虚拟节点[192.168.0.4:111&&VN2]被添加, hash值为704034572
虚拟节点[192.168.0.4:111&&VN3]被添加, hash值为704034573
虚拟节点[192.168.0.4:111&&VN4]被添加, hash值为704034574
[127.0.0.1:1111]的hash值为35944419, 被路由到结点[192.168.0.4:111]
[221.226.0.1:2222]的hash值为973393028, 被路由到结点[192.168.0.2:111]
[10.211.0.1:3333]的hash值为561068530, 被路由到结点[192.168.0.4:111]

从运行结果看，每个点路由到的服务器都是Hash值顺时针离它最近的那个服务器节点，没有任何问题。

通过采取虚拟节点的方法，一个真实结点不再固定在Hash换上的某个点，而是大量地分布在整个Hash环上，这样即使上线、下线服务器，也不会造成整体的负载不均衡。

参考

一致性哈希算法及其在分布式系统中的应用

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

实现一致性哈希java版本

陌生但默默一统江湖的MurmurHash

概要

​ Semaphore当前在多线程环境下被扩放使用，操作系统的信号量是个很重要的概念，在进程控制方面都有应用。Java 并发库的Semaphore可以很轻松完成信号量控制，Semaphore可以控制某个资源可被同时访问的个数，通过 acquire()获取一个许可，如果没有就等待，而release()释放一个许可。用来控制资源同时访问个数

​ 以一个停车场运作为例。假设停车场只有三个车位，一开始三个车位都是空的。这时如果同时来了五辆车，看门人允许其中三辆不受阻碍的进入，然后放下车拦，剩下的车则必须在入口等待，此后来的车也都不得不在入口处等待。这时，有一辆车离开停车场，看门人得知后，打开车拦，放入一辆，如果又离开两辆，则又可以放入两辆，如此往复。

构造函数

Semaphore提供了一个带有boolean参数的构造方法，true代表公平锁，false代表非公平锁，默认实现是非公平锁。

1
2

public Semaphore(int permits); // 创建具有给定许可数的非公平Semaphore
public Semaphore(int permits, boolean fair); //创建具有给定许可数的公平(true)或非公平(false) Semaphore

普通方法

1
2
3
4
5

public void acquire() //从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断
public void acquire(int permits) //从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞，或者线程已被中断
public void release() //释放一个许可，将可用的许可数增加1
public void release(int permits) //释放给定数目的许可，将其返回到信号量
public boolean isFair() //如果此信号量的公平设置为true，则返回 true

停车案例

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

package com.zsr.test.Semaphore;
import java.util.concurrent.Semaphore;
class Car implements Runnable {
  private final Semaphore parkingspace;
  private int carNo;
  /**
   * @param parkingspace
   * @param carNo
   */
  public Car(Semaphore parkingspace, int carNo) {
    this.parkingspace = parkingspace;
    this.carNo = carNo;
  }
  public void run() {
    try {
      parkingspace.acquire();
      parking();
      Thread.sleep(300);
      parkingspace.release();
      leaving();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  private void parking() {
    System.out.println(String.format("%d号车泊车", carNo));
  }
  private void leaving() {
    System.out.println(String.format("%d号车离开车位", carNo));
  }
}
public class ParkingCars {
  private static final int NUMBER_OF_CARS = 5;
  private static final int NUMBER_OF_PARKING_SPACE = 3;
  public static void main(String[] args) throws InterruptedException {
    
    Semaphore parkingSpace = new Semaphore(NUMBER_OF_PARKING_SPACE, true);
    for (int carNo = 1; carNo <= NUMBER_OF_CARS; carNo++) {
      new Thread(new Car(parkingSpace, carNo)).start();
    }
    Thread.sleep(3000);
    /*
     * 输出还有几个可以用的资源数
     */
    System.out.println(parkingSpace.availablePermits() + " 个停车位可以用!");
  }
}

输出结果:

1
2
3
4
5
6
7
8
9
10
11

2号车泊车
1号车泊车
3号车泊车
2号车离开车位
4号车泊车
3号车离开车位
1号车离开车位
5号车泊车
4号车离开车位
5号车离开车位
3 个停车位可以用!