Java technology route
Home Page: https://edagarli.gitbooks.io/java-route/content/
Java成神技术路线，成为Java大神所需要学习的所有相关知识点。其中涉及到一些我写的演示代码。我放在了这里。https://github.com/edagarli/JavaDayAndDay
声明：文章有些内容来自我阅读过的第三方资源，书籍最后我会统一备注引用出处。有任何其他问题，都可以联系我edagarli([email protected])
当然也可以在issues里讨论，我很喜欢大家一起讨论,无论哪方面，只要跟技术有关的都可。
比如: 如何设计一个秒杀系统，等等等等。这样也会成为一个资源共享地。好的内容我会融入这本书中。目前我自己也收集了一部分。
你是一个普通的 Java 对象，你出生在 Eden 区，在 Eden 区有许多和你差不多的小兄弟、小姐妹，可以把 Eden 区当成幼儿园，在这个幼儿园里大家玩了很长时间。Eden 区不能无休止地放你们在里面，所以当年纪稍大，你就要被送到学校去上学，这里假设从小学到高中都称为 Survivor 区。开始的时候你在 Survivor 区里面划分出来的的“From”区，读到高年级了，就进了 Survivor 区的“To”区，中间由于学习成绩不稳定，还经常来回折腾。直到你 18 岁的时候，高中毕业了，该去社会上闯闯了。于是你就去了年老代，年老代里面人也很多。在年老代里，你生活了 20 年 (每次 GC 加一岁)，最后寿终正寝，被 GC 回收。有一点没有提，你在年老代遇到了一个同学，他的名字叫爱德华 (慕光之城里的帅哥吸血鬼)，他以及他的家族永远不会死，那么他们就生活在永生代。
rt.
在多线程程序设计中，经常会遇到一个线程等待一个或多个线程的场景，遇到这样的场景应该如何解决？
如果是一个线程等待一个线程，则可以通过await()和notify()来实现；
如果是一个线程等待多个线程，则就可以使用CountDownLatch和CyclicBarrier来实现比较好的控制。
下面来详细描述下CountDownLatch的应用场景：
例如：百米赛跑：8名运动员同时起跑，由于速度的快慢，肯定有会出现先到终点和晚到终点的情况，而终点有个统计成绩的仪器，当所有选手到达终点时，它会统计所有人的成绩并进行排序，然后把结果发送到汇报成绩的系统。
其实这就是一个CountDownLatch的应用场景：一个线程或多个线程等待其他线程运行达到某一目标后进行自己的下一步工作，而被等待的“其他线程”达到这个目标后继续自己下面的任务。
这个场景中：
那么，如何来通过CountDownLatch来实现上述场景的线程控制和调度呢？
jdk中CountDownLatch类有一个常用的构造方法：CountDownLatch(int count)；
两个常用的方法：await()和countdown()
其 中count是一个计数器中的初始化数字，比如初始化的数字是2，当一个线程里调用了countdown()，则这个计数器就减一，当线程调用了 await()，则这个线程就等待这个计数器变为0，当这个计数器变为0时，这个线程继续自己下面的工作。下面是上述CountDownLatch场景的 实现：
Work类（运动员）：
`
import java.util.concurrent.CountDownLatch;
public class Work implements Runnable {
private int id;
private CountDownLatch beginSignal;
private CountDownLatch endSignal;
public Work(int id, CountDownLatch begin, CountDownLatch end) {
this.id = id;
this.beginSignal = begin;
this.endSignal = end;
}
@OverRide
public void run() {
try {
beginSignal.await();
System.out.println("起跑...");
System.out.println("work" + id + "到达终点");
endSignal.countDown();
System.out.println("work" + id + "继续干其他事情");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
`
Main类（终点统计仪器）：
`
import java.util.concurrent.CountDownLatch;
public class Main {
public static void main(String[] args) {
CountDownLatch begSignal = new CountDownLatch(1);
CountDownLatch endSignal = new CountDownLatch(8);
for (int i = 0; i < 8; i++) {
new Thread(new Work(i, begSignal, endSignal)).start();
}
try {
begSignal.countDown(); //统一起跑
endSignal.await(); //等待运动员到达终点
System.out.println("结果发送到汇报成绩的系统");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
`
下面详细描述下CyclicBarrier的应用场景：
有四个游戏玩家玩游戏，游戏有三个关卡，每个关卡必须要所有玩家都到达后才能允许通关。
其 实这个场景里的玩家中如果有玩家A先到了关卡1，他必须等待其他所有玩家都到达关卡1时才能通过，也就是说线程之间需要互相等待，这和 CountDownLatch的应用场景有区别，CountDownLatch里的线程是到了运行的目标后继续干自己的其他事情，而这里的线程需要等待其 他线程后才能继续完成下面的工作。
jdk中CyclicBarrier类有两个常用的构造方法：
这里的parties也是一个计数器，例如，初始化时parties里的计数是3，于是拥有该CyclicBarrier对象的线程当parties的计数为3时就唤醒，注：这里parties里的计数在运行时当调用CyclicBarrier:await()时,计数就加1，一直加到初始的值
这里的parties与上一个构造方法的解释是一样的，这里需要解释的是第二个入参（Runnable barrierAction）,这个参数是一个实现Runnable接口的类的对象，也就是说当parties加到初始值时就出发barrierAction的内容。
下面来实现上述的应用场景：
Player类（玩家类）
`
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class Player implements Runnable {
private CyclicBarrier cyclicBarrier;
private int id;
public Player(int id, CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
this.id = id;
}
@OverRide
public void run() {
try {
System.out.println("玩家" + id + "正在玩第一关...");
cyclicBarrier.await();
System.out.println("玩家" + id + "进入第二关...");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}
`
GameBarrier类（关卡类，这里控制玩家必须全部到达第一关结束的关口才能进入第二关）
`
import java.util.concurrent.CyclicBarrier;
public class GameBarrier {
public static void main(String[] args) {
CyclicBarrier cyclicBarrier = new CyclicBarrier(4, new Runnable() {
@OverRide
public void run() {
System.out.println("所有玩家进入第二关！");
}
});
for (int i = 0; i < 4; i++) {
new Thread(new Player(i, cyclicBarrier)).start();
}
}
}
`
本文参考翻译自 《Redis Sentinel Documentation》
概述
Redis-Sentinel是Redis官方推荐的高可用性(HA)解决方案，当用Redis做Master-slave的高可用方案时，假如master宕机了，Redis本身(包括它的很多客户端)都没有实现自动进行主备切换，而Redis-sentinel本身也是一个独立运行的进程，它能监控多个master-slave集群，发现master宕机后能进行自懂切换。
它的主要功能有以下几点
Sentinel支持集群
很显然，只使用单个sentinel进程来监控redis集群是不可靠的，当sentinel进程宕掉后(sentinel本身也有单点问题，single-point-of-failure)整个集群系统将无法按照预期的方式运行。所以有必要将sentinel集群，这样有几个好处：
Sentinel版本
Sentinel当前最新的稳定版本称为Sentinel 2(与之前的Sentinel 1区分开来）。随着redis2.8的安装包一起发行。安装完Redis2.8后，可以在redis2.8/src/里面找到Redis-sentinel的启动程序。
强烈建议：
如果你使用的是redis2.6(sentinel版本为sentinel 1)，你最好应该使用redis2.8版本的sentinel 2，因为sentinel 1有很多的Bug，已经被官方弃用，所以强烈建议使用redis2.8以及sentinel 2。
运行Sentinel
运行sentinel有两种方式：
Sentinel的配置
Redis源码包中包含了一个sentinel.conf文件作为sentinel的配置文件，配置文件自带了关于各个配置项的解释。典型的配置项如下所示：
sentinel monitor mymaster 127.0.0.1 6379 2
sentinel down-after-milliseconds mymaster 60000
sentinel failover-timeout mymaster 180000
sentinel parallel-syncs mymaster 1
sentinel monitor resque 192.168.1.3 6380 4
sentinel down-after-milliseconds resque 10000
sentinel failover-timeout resque 180000
sentinel parallel-syncs resque 5
上面的配置项配置了两个名字分别为mymaster和resque的master，配置文件只需要配置master的信息就好啦，不用配置slave的信息，因为slave能够被自动检测到(master节点会有关于slave的消息)。需要注意的是，配置文件在sentinel运行期间是会被动态修改的，例如当发生主备切换时候，配置文件中的master会被修改为另外一个slave。这样，之后sentinel如果重启时，就可以根据这个配置来恢复其之前所监控的redis集群的状态。
接下来我们将一行一行地解释上面的配置项：
sentinel monitor mymaster 127.0.0.1 6379 2
这一行代表sentinel监控的master的名字叫做mymaster,地址为127.0.0.1:6379，行尾最后的一个2代表什么意思呢？我们知道，网络是不可靠的，有时候一个sentinel会因为网络堵塞而误以为一个master redis已经死掉了，当sentinel集群式，解决这个问题的方法就变得很简单，只需要多个sentinel互相沟通来确认某个master是否真的死了，这个2代表，当集群中有2个sentinel认为master死了时，才能真正认为该master已经不可用了。（sentinel集群中各个sentinel也有互相通信，通过gossip协议）。
除了第一行配置，我们发现剩下的配置都有一个统一的格式:
sentinel <option_name> <master_name> <option_value>
接下来我们根据上面格式中的option_name一个一个来解释这些配置项：
不过需要注意的是，这个时候sentinel并不会马上进行failover主备切换，这个sentinel还需要参考sentinel集群中其他sentinel的意见，如果超过某个数量的sentinel也主观地认为该master死了，那么这个master就会被客观地(注意哦，这次不是主观，是客观，与刚才的subjectively down相对，这次是objectively down，简称为ODOWN)认为已经死了。需要一起做出决定的sentinel数量在上一条配置中进行配置。
其他配置项在sentinel.conf中都有很详细的解释。
所有的配置都可以在运行时用命令SENTINEL SET command动态修改。
Sentinel的“仲裁会”
前面我们谈到，当一个master被sentinel集群监控时，需要为它指定一个参数，这个参数指定了当需要判决master为不可用，并且进行failover时，所需要的sentinel数量，本文中我们暂时称这个参数为票数
不过，当failover主备切换真正被触发后，failover并不会马上进行，还需要sentinel中的大多数sentinel授权后才可以进行failover。
当ODOWN时，failover被触发。failover一旦被触发，尝试去进行failover的sentinel会去获得“大多数”sentinel的授权（如果票数比大多数还要大的时候，则询问更多的sentinel)
这个区别看起来很微妙，但是很容易理解和使用。例如，集群中有5个sentinel，票数被设置为2，当2个sentinel认为一个master已经不可用了以后，将会触发failover，但是，进行failover的那个sentinel必须先获得至少3个sentinel的授权才可以实行failover。
如果票数被设置为5，要达到ODOWN状态，必须所有5个sentinel都主观认为master为不可用，要进行failover，那么得获得所有5个sentinel的授权。
配置版本号
为什么要先获得大多数sentinel的认可时才能真正去执行failover呢？
当一个sentinel被授权后，它将会获得宕掉的master的一份最新配置版本号，当failover执行结束以后，这个版本号将会被用于最新的配置。因为大多数sentinel都已经知道该版本号已经被要执行failover的sentinel拿走了，所以其他的sentinel都不能再去使用这个版本号。这意味着，每次failover都会附带有一个独一无二的版本号。我们将会看到这样做的重要性。
而且，sentinel集群都遵守一个规则：如果sentinel A推荐sentinel B去执行failover，B会等待一段时间后，自行再次去对同一个master执行failover，这个等待的时间是通过failover-timeout配置项去配置的。从这个规则可以看出，sentinel集群中的sentinel不会再同一时刻并发去failover同一个master，第一个进行failover的sentinel如果失败了，另外一个将会在一定时间内进行重新进行failover，以此类推。
redis sentinel保证了活跃性：如果大多数sentinel能够互相通信，最终将会有一个被授权去进行failover.
redis sentinel也保证了安全性：每个试图去failover同一个master的sentinel都会得到一个独一无二的版本号。
配置传播
一旦一个sentinel成功地对一个master进行了failover，它将会把关于master的最新配置通过广播形式通知其它sentinel，其它的sentinel则更新对应master的配置。
一个faiover要想被成功实行，sentinel必须能够向选为master的slave发送SLAVE OF NO ONE命令，然后能够通过INFO命令看到新master的配置信息。
当将一个slave选举为master并发送SLAVE OF NO ONE`后，即使其它的slave还没针对新master重新配置自己，failover也被认为是成功了的，然后所有sentinels将会发布新的配置信息。
新配在集群中相互传播的方式，就是为什么我们需要当一个sentinel进行failover时必须被授权一个版本号的原因。
每个sentinel使用##发布/订阅##的方式持续地传播master的配置版本信息，配置传播的##发布/订阅##管道是：sentinel:hello。
因为每一个配置都有一个版本号，所以以版本号最大的那个为标准。
举个栗子：假设有一个名为mymaster的地址为192.168.1.50:6379。一开始，集群中所有的sentinel都知道这个地址，于是为mymaster的配置打上版本号1。一段时候后mymaster死了，有一个sentinel被授权用版本号2对其进行failover。如果failover成功了，假设地址改为了192.168.1.50:9000，此时配置的版本号为2，进行failover的sentinel会将新配置广播给其他的sentinel，由于其他sentinel维护的版本号为1，发现新配置的版本号为2时，版本号变大了，说明配置更新了，于是就会采用最新的版本号为2的配置。
这意味着sentinel集群保证了第二种活跃性：一个能够互相通信的sentinel集群最终会采用版本号最高且相同的配置。
SDOWN和ODOWN的更多细节
sentinel对于不可用有两种不同的看法，一个叫主观不可用(SDOWN),另外一个叫客观不可用(ODOWN)。SDOWN是sentinel自己主观上检测到的关于master的状态，ODOWN需要一定数量的sentinel达成一致意见才能认为一个master客观上已经宕掉，各个sentinel之间通过命令SENTINEL is_master_down_by_addr来获得其它sentinel对master的检测结果。
从sentinel的角度来看，如果发送了PING心跳后，在一定时间内没有收到合法的回复，就达到了SDOWN的条件。这个时间在配置中通过is-master-down-after-milliseconds参数配置。
当sentinel发送PING后，以下回复之一都被认为是合法的：
PING replied with +PONG.
PING replied with -LOADING error.
PING replied with -MASTERDOWN error.
其它任何回复（或者根本没有回复）都是不合法的。
从SDOWN切换到ODOWN不需要任何一致性算法，只需要一个gossip协议：如果一个sentinel收到了足够多的sentinel发来消息告诉它某个master已经down掉了，SDOWN状态就会变成ODOWN状态。如果之后master可用了，这个状态就会相应地被清理掉。
正如之前已经解释过了，真正进行failover需要一个授权的过程，但是所有的failover都开始于一个ODOWN状态。
ODOWN状态只适用于master，对于不是master的redis节点sentinel之间不需要任何协商，slaves和sentinel不会有ODOWN状态。
Sentinel之间和Slaves之间的自动发现机制
虽然sentinel集群中各个sentinel都互相连接彼此来检查对方的可用性以及互相发送消息。但是你不用在任何一个sentinel配置任何其它的sentinel的节点。因为sentinel利用了master的发布/订阅机制去自动发现其它也监控了统一master的sentinel节点。
通过向名为sentinel:hello的管道中发送消息来实现。
同样，你也不需要在sentinel中配置某个master的所有slave的地址，sentinel会通过询问master来得到这些slave的地址的。
每个sentinel通过向每个master和slave的发布/订阅频道sentinel:hello每秒发送一次消息，来宣布它的存在。
每个sentinel也订阅了每个master和slave的频道sentinel:hello的内容，来发现未知的sentinel，当检测到了新的sentinel，则将其加入到自身维护的master监控列表中。
每个sentinel发送的消息中也包含了其当前维护的最新的master配置。如果某个sentinel发现
自己的配置版本低于接收到的配置版本，则会用新的配置更新自己的master配置。
在为一个master添加一个新的sentinel前，sentinel总是检查是否已经有sentinel与新的sentinel的进程号或者是地址是一样的。如果是那样，这个sentinel将会被删除，而把新的sentinel添加上去。
网络隔离时的一致性
redis sentinel集群的配置的一致性模型为最终一致性，集群中每个sentinel最终都会采用最高版本的配置。然而，在实际的应用环境中，有三个不同的角色会与sentinel打交道：
为了考察整个系统的行为我们必须同时考虑到这三个角色。
下面有个简单的例子，有三个主机，每个主机分别运行一个redis和一个sentinel:
+-------------+
| Sentinel 1 | <--- Client A
| Redis 1 (M) |
+-------------+
|
|
+-------------+ | +------------+
| Sentinel 2 |-----+-- / partition / ----| Sentinel 3 | <--- Client B
| Redis 2 (S) | | Redis 3 (M)|
+-------------+ +------------+
在这个系统中，初始状态下redis3是master, redis1和redis2是slave。之后redis3所在的主机网络不可用了，sentinel1和sentinel2启动了failover并把redis1选举为master。
Sentinel集群的特性保证了sentinel1和sentinel2得到了关于master的最新配置。但是sentinel3依然持着的是就的配置，因为它与外界隔离了。
当网络恢复以后，我们知道sentinel3将会更新它的配置。但是，如果客户端所连接的master被网络隔离，会发生什么呢？
客户端将依然可以向redis3写数据，但是当网络恢复后，redis3就会变成redis的一个slave，那么，在网络隔离期间，客户端向redis3写的数据将会丢失。
也许你不会希望这个场景发生：
因为redis采用的是异步复制，在这样的场景下，没有办法避免数据的丢失。然而，你可以通过以下配置来配置redis3和redis1，使得数据不会丢失。
min-slaves-to-write 1
min-slaves-max-lag 10
通过上面的配置，当一个redis是master时，如果它不能向至少一个slave写数据(上面的min-slaves-to-write指定了slave的数量)，它将会拒绝接受客户端的写请求。由于复制是异步的，master无法向slave写数据意味着slave要么断开连接了，要么不在指定时间内向master发送同步数据的请求了(上面的min-slaves-max-lag指定了这个时间)。
Sentinel状态持久化
snetinel的状态会被持久化地写入sentinel的配置文件中。每次当收到一个新的配置时，或者新创建一个配置时，配置会被持久化到硬盘中，并带上配置的版本戳。这意味着，可以安全的停止和重启sentinel进程。
Quorom 机制，是一种分布式系统中常用的，用来保证数据冗余和最终一致性的投票算法，其主要数学**来源于鸽巢原理。
在有冗余数据的分布式存储系统当中，冗余数据对象会在不同的机器之间存放多份拷贝。但是同一时刻一个数据对象的多份拷贝只能用于读或者用于写。
该算法可以保证同一份数据对象的多份拷贝不会被超过两个访问对象读写。
算法来源于[Gifford, 1979][3][1]。 分布式系统中的每一份数据拷贝对象都被赋予一票。每一个操作必须要获得最小的读票数（Vr）或者最小的写票数(Vw）才能读或者写。如果一个系统有V票（意味着一个数据对象有V份冗余拷贝），那么这最小读写票必须满足：
Vr + Vw > V
Vw > V/2
第一条规则保证了一个数据不会被同时读写。当一个写操作请求过来的时候，它必须要获得Vw个冗余拷贝的许可。而剩下的数量是V-Vw 不够Vr，因此不能再有读请求过来了。同理，当读请求已经获得了Vr个冗余拷贝的许可时，写请求就无法获得许可了。
第二条规则保证了数据的串行化修改。一份数据的冗余拷贝不可能同时被两个写请求修改。
算法的好处
在分布式系统中，冗余数据是保证可靠性的手段，因此冗余数据的一致性维护就非常重要。一般而言，一个写操作必须要对所有的冗余数据都更新完成了，才能称为成功结束。比如一份数据在5台设备上有冗余，因为不知道读数据会落在哪一台设备上，那么一次写操作，必须5台设备都更新完成，写操作才能返回。
对于写操作比较频繁的系统，这个操作的瓶颈非常大。Quorum算法可以让写操作只要写完3台就返回。剩下的由系统内部缓慢同步完成。而读操作，则需要也至少读3台，才能保证至少可以读到一个最新的数据。
Quorum的读写最小票数可以用来做为系统在读、写性能方面的一个可调节参数。写票数Vw越大，则读票数Vr越小，这时候系统写的开销就大。反之则写的开销就小。
现在的操作系统都是支持多任务的，多任务可通过多进程或多线程的方式去实现，进程和线程的对比就不在这里说了，在多任务的调度上操作系统采取抢占式和协作式两种方式，抢占式是指操作系统给每个任务一定的执行时间片，在到达这个时间片后如任务仍然未释放对CPU的占用，那么操作系统将强制释放，这是目前多数操作系统采取的方式；协作式是指操作系统按照任务的顺序来分配CPU，每个任务执行过程中除非其主动释放，否则将一直占据CPU，这种方式非常值得注意的是一旦有任务占据CPU不放，会导致其他任务”饿死”的现象，因此操作系统确实不太适合采用这种方式。
说完操作系统多任务的调度方式后，来看看通常程序是如何实现支持高并发的，一种就是典型的基于操作系统提供的多进程或多线程机制，每个任务占据一个进程或一个线程，当任务中有IO等待等动作时，则将进程或线程放入待调度队列中，这种方式是目前大多数程序采取的方式，这种方式的坏处在于如想支持高的并发量，就不得不创建很多的进程或线程，而进程和线程都是要消耗不少系统资源的，另外一方面，进程或线程创建太多后，操作系统需要花费很多的时间在进程或线程的切换上，切换动作需要做状态保持和恢复，这也会消耗掉很多的系统资源；另外一种方式则是每个任务不完全占据一个进程或线程，当任务执行过程中需要进行IO等待等动作时，任务则将其所占据的进程或线程释放，以便其他任务使用这个进程或线程，这种方式的好处在于可以减少所需要的原生的进程或线程数，并且由于操作系统不需要做进程或线程的切换，而是自行来实现任务的切换，其成本会较操作系统切换低，这种方式也就是本文的重点，Coroutine方式，又称协程方式，这种方式在目前的大多数语言中都有支持。
各种语言在实现Coroutine方式的支持时，多数都采用了Actor Model来实现，Actor Model简单来说就是每个任务就是一个Actor，Actor之间通过消息传递的方式来进行交互，而不采用共享的方式，Actor可以看做是一个轻量级的进程或线程，通常在一台4G内存的机器上，创建几十万个Actor是毫无问题的，Actor支持Continuations，即对于如下代码：
Actor
act方法
进行一些处理
创建并执行另外一个Actor
通过消息box阻塞获取另一个Actor执行的结果
继续基于这个结果进行一些处理
在支持Continuations的情况下，可以做到消息box阻塞时并不是进程或线程级的阻塞，而只是Actor本身的阻塞，并且在阻塞时可将所占据的进程或线程释放给其他Actor使用，Actor Model实现最典型的就是erLang了。
对于Java应用而言，传统方式下为了支持高并发，由于一个线程只能用于处理一个请求，即使是线程中其实有很多IO中断、锁等待也同样如此，因此通常的做法是通过启动很多的线程来支撑高并发，但当线程过多时，就造成了CPU需要消耗不少的时间在线程的切换上，从而出现瓶颈，按照上面对Coroutine的描述，Coroutine的方式理论上而言能够大幅度的提升Java应用所能支撑的并发量。
Java尚不能从语言层次上支持Coroutine，在java中使用协程，可以使用协程框架，Kilim就是一种比较流行的协程框架。
Kilim是由剑桥的两位博士开发的一个用于在Java中使用Coroutine的框架，Kilim基于Java语法。( 具体了解可看这个，http://www.malhar.net/sriram/kilim/ 或 http://www.ibm.com/developerworks/cn/java/j-javadev2-7.html)
经测试，当线程数增加时，系统时耗激增。由于操作系统本地线程数量的限制，无法正常运行线程数8000和10000的测试代码。而使用协程的方式，其增长曲线显得十分平缓。使用协程，可以让系统以更低的成本，支持更高的并行度。
总结而言，采用Coroutine方式可以很好的绕开需要启动太多线程来支撑高并发出现的瓶颈，提高Java应用所能支撑的并发量，但在开发模式上也会带来变化，并且需要特别注意不能造成线程被阻塞的现象，从开发易用和透明迁移现有Java应用两个角度而言目前Coroutine方式还有很多不足，但相信随着越来越多的人在Java中使用Coroutine，其易用性必然是能够得到提升的。
相关资料:
http://en.wikipedia.org/wiki/Computer_multitasking
http://en.wikipedia.org/wiki/Coroutine
http://en.wikipedia.org/wiki/Actor_model
http://en.wikipedia.org/wiki/Continuation
http://lamp.epfl.ch/~phaller/doc/haller07coord.pdf
http://www.scala-lang.org/sites/default/files/odersky/jmlc06.pdf
http://www.malhar.net/sriram/kilim/kilim_ecoop08.pdf
http://lamp.epfl.ch/~phaller/doc/ScalaActors.pdf
http://blog.csdn.net/kobejayandy/article/details/41412787
http://www.blogjava.net/BlueDavy/archive/2010/01/28/311148.html
互联网技术联盟（ITA1024）于4月开始推出大数据技术专题月，每周安排一场线上精英群分享和一整天10大议题的线下1024大数据技术峰会。
http://gitbook.cn/books/57107c8976dc085d7a00cb04/index.html
https://zhuanlan.zhihu.com/p/20872901
API 调用次数限制实现
在开发接口服务器的过程中，为了防止客户端对于接口的滥用，保护服务器的资源， 通常来说我们会对于服务器上的各种接口进行调用次数的限制。比如对于某个 用户，他在一个时间段（interval）内，比如 1 分钟，调用服务器接口的次数不能够 大于一个上限（limit），比如说 100 次。如果用户调用接口的次数超过上限的话，就 直接拒绝用户的请求，返回错误信息。
最开始的想法
对于实现方法的第一印象，大概是，给每个用户一个配额，次数为 Q， 这个配额在用户第一次调用接口的时候分配给用户。然后在接下去的 P 时间段 内，如果用户访问 API 的次数大于 Q，就开始拒绝用户的调用请求。然后，这个 配额，在 P 时间之后，配额会被重置回 Q。
大概的伪代码如下：
can_access(identity):
limit_counter = get(identity)
if limiter_counter exists and
limiter_counter.timestamp - CURRENT_TIME < LIMIT_INTERVAL and
limit_counter >= limit:
return false
else
if limiter_counter is nil or
(limit_counter exists and limiter_counter.timestamp - CURRENT_TIME > LIMT_INTERVAL):
put(identity, new LimiterCounter(1, CURRENT_TIME))
else
put(identity, limiter_counter.increment())
end
return true
end
redis 的 INCR 命令可以比较简单地实现这种方法，在 INCR 的文档页下面介绍了如何使用 INCR 命令实现 Rate Limtier。
这种实现方法，仔细想来，存在一个缺陷，就是，用户可以在一个时间段的末尾发起 Q 次请求，然后在下一个时间段的开始又 发起 Q 次请求，这样，一个用户可以在很短的时间之内发起 2Q 次请求。
可能普通用户不会刻意这么去做，但如果真的出现这种情况的时候，服务器会承受正常情况下两倍的负载， 这并不是我们所希望看见的。而且如果服务器被攻击的话，这种的缺陷，还是很可能会被利用的。
Token Bucket 算法
作为一个程序员，当我对于一个问题没有头绪时，可以
或者可以
简单的 Google 之后，在 StackOverflow 上面了解到 Token Bucket 算法， 这个算法通常用在计算机的网络设备上面，一般用于限制流量，很符合我所需要解决的问题。
在这个算法当中：
所有的流量在放行之前需要获取一定量的 token；
所有的 token 存放在一个 bucket（桶）当中，每 1/r 秒，都会往这个 bucket 当中加入一个 token；
bucket 有最大容量（capacity or limit），在 bucket 中的 token 数量等于最大容量，而且没有 token 消耗时，新的额外的 token 会被抛弃。
简单地来看，可以将这个算法类比成有个水龙头在往水桶中放水，然后不断地有水瓢到这个水桶中打水去浇花，如果水桶的水满了，那么 水就从水桶中溢出了。
在我们的问题领域，要将流量换成一个请求。当一个请求到达服务器的时候，首先需要根据请求的各种信息，确定其需要获取哪个 bucket 的 Token，因为服务器一般会有多种限制流量的策略进行组合。
举一些例子：
对于每个登录过的用户，服务器规定 10 秒内，用户的请求次数不能超过 200 次；而且，1 小时内，用户的请求次数不能超过 5000 次；并且，1 天内， 用户的请求次数不能超过 20000 次。这样，对于每个用户都需要设置三个 bucket。
另外服务器还规定服务器所有的接口在 10 秒内，请求次数不能超过 100000 次，这种情况下，所有用户会共享一个全服务器的 token bucket。
还有可能根据 IP 进行限制，这样 bucket 就需要根据 IP 地址进行创建。
接着，这个请求去对应的 token bucket 获取允许通行的 token，如果没有获取到 token，服务器最好的做法是直接返回流量超过限制的响应（429）。 如果获取到相应的 token，那么就对于请求给予放行。
初步的实现的想法
初步实现的想法很简单，首先，对于每个 Bucket 设置一个定时器，每过一个间隔，就往这个 Bucket 当中加入一些 Token，然后用户 获取一个 Token 之后，就将 Bucket 中的 Token 数量减一。这个实现，是在看到这个算法的时候，比较容易想到的一个方法。然而， 稍微仔细地考虑一下，就知道这个实现手段在现实当中基本上是属于没法用的实现。原因在于，这种实现算法需要给每个 Bucket 添加 一个定时器，而一个定时器就是一条线程。那么在你的服务器上，光是分配给定时器的线程就需要和你的用户数量是一个量级的， 几万几十万条线程在服务器上运行，是完全是脱离了实际情况的。
所以这个简单的实现方法，在稍微考虑之后，就可以排除了。
另一种的实现的办法
于是又开始寻找另外的实现方法，搜索资料的时候，发现 Guava 库当中也有一个 RateLimiter，其作用也是 用来进行限流，于是阅读了 RateLimiter 的源代码，查看一些 Google 的人是如何实现 Token Bucket 算法的。
RateLimiter 和 SmoothRateLimiter
// com.google.common.util.concurrent.SmoothRateLimiter
private void resync(long nowMicros) {
// if nextFreeTicket is in the past, resync to now
if (nowMicros > nextFreeTicketMicros) {
storedPermits = min(maxPermits,
storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);
nextFreeTicketMicros = nowMicros;
}
}
在 resync 方法中的这句代码 storedPermits = min(maxPermits, storedPermits+ (nowMicros - nextFreeTicketMicros)/stableIntervalMicros); 就是 RateLimiter 中计算 Token 数量的方法。没有使用计时器，而是使用时间戳的方式计算。这个做法给足了 信息。我们可以在 Bucket 中存放现在的 Token 数量，然后存储上一次补充 Token 的时间戳，当用户下一次请求获取一个 Token 的时候， 根据此时的时间戳，计算从上一个时间戳开始，到现在的这个时间点所补充的所有 Token 数量，加入到 Bucket 当中。
这种实现方法有几个优势：
首先, 避免了给每一个 Bucket 设置一个定时器这种笨办法，
第二，数据结构需要的内存量很小，只需要储存 Bucket 中剩余的 Token 量以及上次补充 Token 的时间戳就可以了；
第三，只有在用户访问的时候，才会计算 Token 补充量，对于系统的计算资源占用量也较小。
确定和实现方法之后，就可以开始实现这个算法了。首先要考虑的是 Bucket 存放在哪里？虽然 Bucket 占用内存的数量 很小，假设一个 Bucket 的大小为 20 个字节，如果需要储存一百万个 Bucket 就需要使用 20M 的内存。而且，Bucket 从 一定意义上属于缓存数据，因为如果用户长期不使用这个 Bucket 的话，应该能够自动失效。从上面的分析，自然地，我想到 将 Bucket 放在 Redis 当中，每个 Bucket 使用一个 Hash 存放（HSET）， 并且支持在一段时间之后，使 Bucket 失效（TTL）。
于是有了第一版的代码， RateLimiter
public boolean access(String userId) {
String key = genKey(userId);
try (Jedis jedis = jedisPool.getResource()) {
Map<String, String> counter = jedis.hgetAll(key);
if (counter.size() == 0) {
TokenBucket tokenBucket = new TokenBucket(System.currentTimeMillis(), limit - 1);
jedis.hmset(key, tokenBucket.toHash());
return true;
} else {
TokenBucket tokenBucket = TokenBucket.fromHash(counter);
long lastRefillTime = tokenBucket.getLastRefillTime();
long refillTime = System.currentTimeMillis();
long intervalSinceLast = refillTime - lastRefillTime;
long currentTokensRemaining;
if (intervalSinceLast > intervalInMills) {
currentTokensRemaining = limit;
} else {
long grantedTokens = (long) (intervalSinceLast / intervalPerPermit);
System.out.println(grantedTokens);
currentTokensRemaining = Math.min(grantedTokens + tokenBucket.getTokensRemaining(), limit);
}
tokenBucket.setLastRefillTime(refillTime);
assert currentTokensRemaining >= 0;
if (currentTokensRemaining == 0) {
tokenBucket.setTokensRemaining(currentTokensRemaining);
jedis.hmset(key, tokenBucket.toHash());
return false;
} else {
tokenBucket.setTokensRemaining(currentTokensRemaining - 1);
jedis.hmset(key, tokenBucket.toHash());
return true;
}
}
}
}
上面的方法是最初的实现方法，对于每一个 Token Bucket，在 Redis 上面，使用一个 Hash 进行表示，一个 Token Bucket 有 lastRefillTime 表示最后一次补充 Token 的时间，tokensRemaining 则表示 Bucket 中的剩余 Token 数量，access() 方法大致的步骤为：
当一个请求 Token进入 access() 方法后，先计算计算该请求的 Token Bucket 的 key；
如果这个 Token Bucket 在 Redis 中不存在，那么就新建一个 Token Bucket，然后设置该 Bucket 的 Token 数量为最大值减一(去掉了这次请求获取的 Token）。 在初始化 Token Bucket 的时候将 Token 数量设置为最大值这一点在后面还有讨论；
如果这个 Token Bucket 在 Redis 中存在，而且其上一次加入 Token 的时间到现在时间的时间间隔大于 Token Bucket 的 interval，那么也将 Bucket 的 Token 值重置为最大值减一；
如果 Token Bucket 上次加入 Token 的时间到现在时间的时间间隔没有大于 interval，那么就计算这次需要补充的 Token 数量，将补充过后的 Token 数量更新到 Token Bucket 中。
这个方法在单线程的条件下面，可以比较好地满足需求，但是在多线程的条件下面，是会出现 race condition，如下面的图所示。
更好一点的实现方法
上面的 race condition 出现的原因是多个线程对于 Token Bucket 进行写操作，当遇到 race condition 的时候， 我们通常使用锁的方式解决这个冲突。
对于上面这个情况，我们需要使用锁保护的资源就是相应的 Token Bucket。如果使用锁的方式实现，需要对每一个 Token Bucket 附加一个锁，当多个线程并发地读写 Token Bucket 的时候，需要先获取这个锁的控制权，然后 对于 Token Bucket 进行修改，然后更新到 Redis 中。
虽然使用在业务代码当中可以实现这个逻辑，但 Redis 提供了一种我个人感觉更好的方法来实现同样的上锁机制。 EVAL 和 EVALSHA， 使用 lua 脚本的方式执行命令，这个脚本整个的操作会被当成一个原子操作在 Redis 上面执行
我们可以将原本的 Java 实现转写成 lua 脚本，将要本要做的 Token Bucket 的读写操作放在这个脚本当中， 让 Redis 去保证这个操作的原子性。 大致的实现放在这里rate_limiter.lua 和LuaRateLimiter.java。
查漏补缺
在上面一个版本的 LuaRateLimiter 当中，我自己发现的问题有两个：
当用户首次请求 Token 或者长时间没有请求 Token 的情况下，首次进行 Token 请求，Bucket 此时所持有的 Token 数量应该是可以设置的，而不是一个简单的 Token Bucket 的容量最大值。因为我们的算法在 Token Bucket 被激活（第一次使用，或者间隔时间很长后使用）之后，会不断地往里面继续添加 Token（在用户请求的时候）， 这样在一个 interval 之内，用户所能够使用最大 Token 数量 M 等于初始的 Token 数量 I 加上 Bucket 的 Token 容量最大值 C。
LuaRateLimiter 承担了两个任务，第一，控制请求从 Token Bucket 中获取 Token；第二，指定 Token Bucket 的 限制规则。在 OOP 编程中，我们一般遵循 Single Responsibility Principle， 而且希望对于一个类来说是高内聚，低耦合的。 在这里，LuaRateLimiter 的两个责任，没有很明显的相互依赖关系，Token Bucket 的限制规则可以独立于控制请求的过程而存在， 而请求控制过程依赖到的是使用 Token Bucket 的限制规则和请求的身份生成 Token Bucket 的 key。在我看来，这构成了很好的理由 将 LuaRateLimiter 拆分成两个类 RateLimitPolicy 和 LuaRateLimiter，来分别承担这两个任务。
经过改版之后的代码在这里
对于第一个问题，我现在做的改动比较有限，在 RateLimitPolicy 当中添加一个 maxBurstTime，然后计算 Bucket 激活的时候 初始的 Token 容量。程序创建一个 RateLimitPolicy 的时候，需要指定这个 maxBurstTime。关于这个初始容量的设置与计算 可以进一步参考 Guava 的 SmoothRateLimiter 中的文档和代码。
第二问题就是比较简单的重构，在实现当中，RateLimitPolicy 是一个抽象类，有一个 PerUserRateLimitPolicy，这个 规则通过 genBucketKey() 方法，对于每个用户都返回一个不同的 bucket key，从而使不同的用户使用不同的 Token Bucket。 另外，如果有别的限制策略，可以通过实现不同的 RateLimitPolicy 来完成（genBucketKey() 方法的定义还可以进一步地优化）。
现在这个版本的实现，不算特别成熟，但如果要求不是特别高的话，也是一个可用的实现方案，所以暂时就先写成这样， 如果实际应用中发现问题，可以进一步地优化。
测试
测试的代码放在LuaRateLimiterTest 中，暂时只是简单地进行了功能上面的测试。
另外可选的实现
除了文章介绍的方法，还有另外一些其他的方法可以实现，阅读参考资料中的 第 2，4，5 篇的文章中介绍的实现方法来完成。
参考资料
Token Bucket
Redis Incr
Redis Eval
Better Rate Limiting With Redis Sorted Sets
Intro to rate limiting with Redis part1 and part2
Guava RateLimiter and Guava SmoothRateLimiter， 特别推荐 SmoothRateLimiter 中的文档部分
Lua Reference，redis 中使用 lua 5.1
Single Responsibility Principle
High Cohesion, Loose Coupling
Redis为单进程单线程模式，采用队列模式将并发访问变为串行访问。Redis本身没有锁的概念，Redis对于多个客户端连接并不存在竞争，但是在Jedis客户端对Redis进行并发访问时会发生连接超时、数据转换错误、阻塞、客户端关闭连接等问题，这些问题均是由于客户端连接混乱造成。对此有2种解决方法：
1.客户端角度，为保证每个客户端间正常有序与Redis进行通信，对连接进行池化，同时对客户端读写Redis操作采用内部锁synchronized。
2.服务器角度，利用setnx实现锁。
对于第一种，需要应用程序自己处理资源的同步，可以使用的方法比较通俗，可以使用synchronized也可以使用lock；第二种需要用到Redis的setnx命令，但是需要注意一些问题。
SETNX命令（SET if Not eXists）
语法：
SETNX key value
功能：
将 key 的值设为 value ，当且仅当 key 不存在；若给定的 key 已经存在，则 SETNX 不做任何动作。
时间复杂度：
O(1)
返回值：
设置成功，返回 1 。
设置失败，返回 0 。
模式：将 SETNX 用于加锁(locking)
SETNX 可以用作加锁原语(locking primitive)。比如说，要对关键字(key) foo 加锁，客户端可以尝试以下方式：
SETNX lock.foo <current Unix time + lock timeout + 1>
如果 SETNX 返回 1 ，说明客户端已经获得了锁， key 设置的unix时间则指定了锁失效的时间。之后客户端可以通过 DEL lock.foo 来释放锁。
如果 SETNX 返回 0 ，说明 key 已经被其他客户端上锁了。如果锁是非阻塞(non blocking lock)的，我们可以选择返回调用，或者进入一个重试循环，直到成功获得锁或重试超时(timeout)。
但是已经证实仅仅使用SETNX加锁带有竞争条件，在特定的情况下会造成错误。
处理死锁(deadlock)
上面的锁算法有一个问题：如果因为客户端失败、崩溃或其他原因导致没有办法释放锁的话，怎么办？
这种状况可以通过检测发现——因为上锁的 key 保存的是 unix 时间戳，假如 key 值的时间戳小于当前的时间戳，表示锁已经不再有效。
但是，当有多个客户端同时检测一个锁是否过期并尝试释放它的时候，我们不能简单粗暴地删除死锁的 key ，再用 SETNX 上锁，因为这时竞争条件(race condition)已经形成了：
C1 和 C2 读取 lock.foo 并检查时间戳， SETNX 都返回 0 ，因为它已经被 C3 锁上了，但 C3 在上锁之后就崩溃(crashed)了。
C1 向 lock.foo 发送 DEL 命令。
C1 向 lock.foo 发送 SETNX 并成功。
C2 向 lock.foo 发送 DEL 命令。
C2 向 lock.foo 发送 SETNX 并成功。
出错：因为竞争条件的关系，C1 和 C2 两个都获得了锁。
幸好，以下算法可以避免以上问题。来看看我们聪明的 C4 客户端怎么办：
C4 向 lock.foo 发送 SETNX 命令。
因为崩溃掉的 C3 还锁着 lock.foo ，所以 Redis 向 C4 返回 0 。
C4 向 lock.foo 发送 GET 命令，查看 lock.foo 的锁是否过期。如果不，则休眠(sleep)一段时间，并在之后重试。
另一方面，如果 lock.foo 内的 unix 时间戳比当前时间戳老，C4 执行以下命令：
GETSET lock.foo <current Unix timestamp + lock timeout + 1>
因为 GETSET 的作用，C4 可以检查看 GETSET 的返回值，确定 lock.foo 之前储存的旧值仍是那个过期时间戳，如果是的话，那么 C4 获得锁。
如果其他客户端，比如 C5，比 C4 更快地执行了 GETSET 操作并获得锁，那么 C4 的 GETSET 操作返回的就是一个未过期的时间戳(C5 设置的时间戳)。C4 只好从第一步开始重试。
注意，即便 C4 的 GETSET 操作对 key 进行了修改，这对未来也没什么影响。
这里假设锁key对应的value没有实际业务意义，否则会有问题，而且其实其value也确实不应该用在业务中。
为了让这个加锁算法更健壮，获得锁的客户端应该常常检查过期时间以免锁因诸如 DEL 等命令的执行而被意外解开，因为客户端失败的情况非常复杂，不仅仅是崩溃这么简单，还可能是客户端因为某些操作被阻塞了相当长时间，紧接着 DEL 命令被尝试执行(但这时锁却在另外的客户端手上)。
GETSET命令
语法：
GETSET key value
功能：
将给定 key 的值设为 value ，并返回 key 的旧值(old value)。当 key 存在但不是字符串类型时，返回一个错误。
时间复杂度：
O(1)
返回值：
返回给定 key 的旧值；当 key 没有旧值时，也即是， key 不存在时，返回 nil 。
其它资源
Redis学习-并发访问，mysql通信 http://blog.csdn.net/hpb21/article/details/7893013
基于Redis实现分布式锁 http://blog.csdn.net/ugg/article/details/41894947
谈谈Redis的SETNX http://huoding.com/2015/09/14/463
静态编译优化和动态编译优化最大的不同是他们在编译时所得到的信息量的不同。静态编译在运行程序之前就把所有的执行代码编译完，这时编译器所接受的编译信息量是不够多的。比如说：某个函数是否是大量地被调用了，函数的实参是不是一直是一个常数，等等。
动态编译之于静态编译，缺点是它需要即时编译代码，但是有一个优点---编译器可以获得静态编译期所没有的信息。比如：通过运行时的profiling可以知道哪些函数是被大量使用的。在哪些execution path上哪些函数的参数一直都没有变，等等。不要小看这些信息，当即时编译器了解这些信息之后可以在短时间内编译出比静态编译器更优质的二进制码。举例来说，一般程序也遵循90-10原则，即运行时的90%里计算机是在处理其中10%的代码，寻找到这些执行热点代码进行深度优化能得到比静态编译更好的性能（因为已知更多信息量）。
然而现实是：即时编译的开销非常大，暂时还不能超越静态编译的总体性能。不过，一个动态语言（如JAVA，Python）有着静态语言（如C++）所没有的各种优势，必然是将来程序语言发展的方向。伴随着强大的需求，即时编译器在将来也会更加强大。
正如每个Java文档所描述的那样，CountDownLatch是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作执行完后再执行。在Java并发中，countdownlatch的概念是一个常见的面试题，所以一定要确保你很好的理解了它。在这篇文章中，我将会涉及到在Java并发编 程中跟CountDownLatch相关的以下几点：
目录
CountDownLatch是什么？
CountDownLatch如何工作？
在实时系统中的应用场景
应用范例
常见的面试题
CountDownLatch是什么
CountDownLatch是在java1.5被引入的，跟它一起被引入的并发工具类还有CyclicBarrier、Semaphore、ConcurrentHashMap和BlockingQueue，它们都存在于java.util.concurrent包下。CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。
CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。
CountDownLatch的伪代码如下所示：
//Main thread start
//Create CountDownLatch for N threads
//Create and start N threads
//Main thread wait on latch
//N threads completes there tasks are returns
//Main thread resume execution
CountDownLatch如何工作
CountDownLatch.java类中定义的构造函数：
//Constructs a CountDownLatch initialized with the given count. public void CountDownLatch(int count) {...}
构造器中的计数值（count）实际上就是闭锁需要等待的线程数量。这个值只能被设置一次，而且CountDownLatch没有提供任何机制去重新设置这个计数值。
与CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。
其他N 个线程必须引用闭锁对象，因为他们需要通知CountDownLatch对象，他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的；每调用一次这个方法，在构造函数中初始化的count值就减1。所以当N个线程都调 用了这个方法，count的值等于0，然后主线程就能通过await()方法，恢复执行自己的任务。
在实时系统中的使用场景
让我们尝试罗列出在java实时系统中CountDownLatch都有哪些使用场景。我所罗列的都是我所能想到的。如果你有别的可能的使用方法，请在留言里列出来，这样会帮助到大家。
实现最大的并行性：有时我们想同时启动多个线程，实现最大程度的并行性。例如，我们想测试一个单例类。如果我们创建一个初始计数为1的CountDownLatch，并让所有线程都在这个锁上等待，那么我们可以很轻松地完成测试。我们只需调用 一次countDown()方法就可以让所有的等待线程同时恢复执行。
开始执行前等待n个线程完成各自任务：例如应用程序启动类要确保在处理用户请求前，所有N个外部系统已经启动和运行了。
死锁检测：一个非常方便的使用场景是，你可以使用n个线程访问共享资源，在每次测试阶段的线程数目是不同的，并尝试产生死锁。
CountDownLatch使用例子
在这个例子中，我模拟了一个应用程序启动类，它开始时启动了n个线程类，这些线程将检查外部系统并通知闭锁，并且启动类一直在闭锁上等待着。一旦验证和检查了所有外部服务，那么启动类恢复执行。
BaseHealthChecker.java：这个类是一个Runnable，负责所有特定的外部服务健康的检测。它删除了重复的代码和闭锁的中心控制代码。
`
public abstract class BaseHealthChecker implements Runnable {
private CountDownLatch _latch;
private String _serviceName;
private boolean _serviceUp;
//Get latch object in constructor so that after completing the task, thread can countDown() the latch
public BaseHealthChecker(String serviceName, CountDownLatch latch)
{
super();
this._latch = latch;
this._serviceName = serviceName;
this._serviceUp = false;
}
@Override
public void run() {
try {
verifyService();
_serviceUp = true;
} catch (Throwable t) {
t.printStackTrace(System.err);
_serviceUp = false;
} finally {
if(_latch != null) {
_latch.countDown();
}
}
}
public String getServiceName() {
return _serviceName;
}
public boolean isServiceUp() {
return _serviceUp;
}
//This methos needs to be implemented by all specific service checker
public abstract void verifyService();
}
`
NetworkHealthChecker.java：这个类继承了BaseHealthChecker，实现了verifyService()方法。DatabaseHealthChecker.java和CacheHealthChecker.java除了服务名和休眠时间外，与NetworkHealthChecker.java是一样的。
`
public class NetworkHealthChecker extends BaseHealthChecker
{
public NetworkHealthChecker (CountDownLatch latch) {
super("Network Service", latch);
}
@Override
public void verifyService()
{
System.out.println("Checking " + this.getServiceName());
try
{
Thread.sleep(7000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
System.out.println(this.getServiceName() + " is UP");
}
}
`
ApplicationStartupUtil.java：这个类是一个主启动类，它负责初始化闭锁，然后等待，直到所有服务都被检测完。
`
public class ApplicationStartupUtil
{
//List of service checkers
private static List _services;
//This latch will be used to wait on
private static CountDownLatch _latch;
private ApplicationStartupUtil()
{
}
private final static ApplicationStartupUtil INSTANCE = new ApplicationStartupUtil();
public static ApplicationStartupUtil getInstance()
{
return INSTANCE;
}
public static boolean checkExternalServices() throws Exception
{
//Initialize the latch with number of service checkers
_latch = new CountDownLatch(3);
//All add checker in lists
_services = new ArrayList<BaseHealthChecker>();
_services.add(new NetworkHealthChecker(_latch));
_services.add(new CacheHealthChecker(_latch));
_services.add(new DatabaseHealthChecker(_latch));
//Start service checkers using executor framework
Executor executor = Executors.newFixedThreadPool(_services.size());
for(final BaseHealthChecker v : _services)
{
executor.execute(v);
}
//Now wait till all services are checked
_latch.await();
//Services are file and now proceed startup
for(final BaseHealthChecker v : _services)
{
if( ! v.isServiceUp())
{
return false;
}
}
return true;
}
}
`
现在你可以写测试代码去检测一下闭锁的功能了。
`
public class Main {
public static void main(String[] args)
{
boolean result = false;
try {
result = ApplicationStartupUtil.checkExternalServices();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("External services validation completed !! Result was :: "+ result);
}
}
`
Output in console:
Checking Network Service
Checking Cache Service
Checking Database Service
Database Service is UP
Cache Service is UP
Network Service is UP
External services validation completed !! Result was :: true
常见面试题
可以为你的下次面试准备以下一些CountDownLatch相关的问题：
解释一下CountDownLatch概念?
CountDownLatch 和CyclicBarrier的不同之处?
给出一些CountDownLatch使用的例子?
CountDownLatch 类中主要的方法?
Java 虚拟机有自己完善的硬件架构, 如处理器、堆栈、寄存器等，还具有相应的指令系统。JVM 屏蔽了与具体操作系统平台相关的信息，使得 Java 程序只需生成在 Java 虚拟机上运行的目标代码 (字节码), 就可以在多种平台上不加修改地运行。Java 虚拟机在执行字节码时，实际上最终还是把字节码解释成具体平台上的机器指令执行。
注意：本文仅针对 JDK7、HotSPOT Java 虚拟机，对于 JDK8 引入的 JVM 新特性及其他 Java 虚拟机，本文不予关注。
我们以一个例子开始这篇文章。假设你是一个普通的 Java 对象，你出生在 Eden 区，在 Eden 区有许多和你差不多的小兄弟、小姐妹，可以把 Eden 区当成幼儿园，在这个幼儿园里大家玩了很长时间。Eden 区不能无休止地放你们在里面，所以当年纪稍大，你就要被送到学校去上学，这里假设从小学到高中都称为 Survivor 区。开始的时候你在 Survivor 区里面划分出来的的“From”区，读到高年级了，就进了 Survivor 区的“To”区，中间由于学习成绩不稳定，还经常来回折腾。直到你 18 岁的时候，高中毕业了，该去社会上闯闯了。于是你就去了年老代，年老代里面人也很多。在年老代里，你生活了 20 年 (每次 GC 加一岁)，最后寿终正寝，被 GC 回收。有一点没有提，你在年老代遇到了一个同学，他的名字叫爱德华 (慕光之城里的帅哥吸血鬼)，他以及他的家族永远不会死，那么他们就生活在永生代。
如何将新对象预留在年轻代
众所周知，由于 Full GC 的成本远远高于 Minor GC，因此某些情况下需要尽可能将对象分配在年轻代，这在很多情况下是一个明智的选择。虽然在大部分情况下，JVM 会尝试在 Eden 区分配对象，但是由于空间紧张等问题，很可能不得不将部分年轻对象提前向年老代压缩。因此，在 JVM 参数调优时可以为应用程序分配一个合理的年轻代空间，以最大限度避免新对象直接进入年老代的情况发生。清单 1 所示代码尝试分配 4MB 内存空间，观察一下它的内存使用情况。
清单 1. 相同大小内存分配
public class PutInEden {
public static void main(String[] args){
byte[] b1,b2,b3,b4;//定义变量
b1=new byte[1024*1024];//分配 1MB 堆空间，考察堆空间的使用情况
b2=new byte[1024*1024];
b3=new byte[1024*1024];
b4=new byte[1024*1024];
}
}
使用 JVM 参数-XX:+PrintGCDetails -Xmx20M -Xms20M 运行清单 1 所示代码，输出如清单 2 所示。
清单 2. 清单 1 运行输出
[GC [DefNew: 5504K->640K(6144K), 0.0114236 secs] 5504K->5352K(19840K),
0.0114595 secs] [Times: user=0.02 sys=0.00, real=0.02 secs]
[GC [DefNew: 6144K->640K(6144K), 0.0131261 secs] 10856K->10782K(19840K),
0.0131612 secs] [Times: user=0.02 sys=0.00, real=0.02 secs]
[GC [DefNew: 6144K->6144K(6144K), 0.0000170 secs][Tenured: 10142K->13695K(13696K),
0.1069249 secs] 16286K->15966K(19840K), [Perm : 376K->376K(12288K)],
0.1070058 secs] [Times: user=0.03 sys=0.00, real=0.11 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0302067 secs] 19839K->19595K(19840K),
[Perm : 376K->376K(12288K)], 0.0302635 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0311986 secs] 19839K->19839K(19840K),
[Perm : 376K->376K(12288K)], 0.0312515 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0358821 secs] 19839K->19825K(19840K),
[Perm : 376K->371K(12288K)], 0.0359315 secs] [Times: user=0.05 sys=0.00, real=0.05 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0283080 secs] 19839K->19839K(19840K),
[Perm : 371K->371K(12288K)], 0.0283723 secs] [Times: user=0.02 sys=0.00, real=0.01 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0284469 secs] 19839K->19839K(19840K),
[Perm : 371K->371K(12288K)], 0.0284990 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0283005 secs] 19839K->19839K(19840K),
[Perm : 371K->371K(12288K)], 0.0283475 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0287757 secs] 19839K->19839K(19840K),
[Perm : 371K->371K(12288K)], 0.0288294 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0288219 secs] 19839K->19839K(19840K),
[Perm : 371K->371K(12288K)], 0.0288709 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0293071 secs] 19839K->19839K(19840K),
[Perm : 371K->371K(12288K)], 0.0293607 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 13695K->13695K(13696K), 0.0356141 secs] 19839K->19838K(19840K),
[Perm : 371K->371K(12288K)], 0.0356654 secs] [Times: user=0.01 sys=0.00, real=0.03 secs]
Heap
def new generation total 6144K, used 6143K [0x35c10000, 0x362b0000, 0x362b0000)
eden space 5504K, 100% used [0x35c10000, 0x36170000, 0x36170000)
from space 640K, 99% used [0x36170000, 0x3620fc80, 0x36210000)
to space 640K, 0% used [0x36210000, 0x36210000, 0x362b0000)
tenured generation total 13696K, used 13695K [0x362b0000, 0x37010000, 0x37010000)
the space 13696K, 99% used [0x362b0000, 0x3700fff8, 0x37010000, 0x37010000)
compacting perm gen total 12288K, used 371K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706cd20, 0x3706ce00, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
清单 2 所示的日志输出显示年轻代 Eden 的大小有 5MB 左右。分配足够大的年轻代空间，使用 JVM 参数-XX:+PrintGCDetails -Xmx20M -Xms20M-Xmn6M 运行清单 1 所示代码，输出如清单 3 所示。
清单 3. 增大 Eden 大小后清单 1 运行输出
[GC [DefNew: 4992K->576K(5568K), 0.0116036 secs] 4992K->4829K(19904K),
0.0116439 secs] [Times: user=0.02 sys=0.00, real=0.02 secs]
[GC [DefNew: 5568K->576K(5568K), 0.0130929 secs] 9821K->9653K(19904K),
0.0131336 secs] [Times: user=0.02 sys=0.00, real=0.02 secs]
[GC [DefNew: 5568K->575K(5568K), 0.0154148 secs] 14645K->14500K(19904K),
0.0154531 secs] [Times: user=0.00 sys=0.01, real=0.01 secs]
[GC [DefNew: 5567K->5567K(5568K), 0.0000197 secs][Tenured: 13924K->14335K(14336K),
0.0330724 secs] 19492K->19265K(19904K), [Perm : 376K->376K(12288K)],
0.0331624 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 14335K->14335K(14336K), 0.0292459 secs] 19903K->19902K(19904K),
[Perm : 376K->376K(12288K)], 0.0293000 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 14335K->14335K(14336K), 0.0278675 secs] 19903K->19903K(19904K),
[Perm : 376K->376K(12288K)], 0.0279215 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured: 14335K->14335K(14336K), 0.0348408 secs] 19903K->19889K(19904K),
[Perm : 376K->371K(12288K)], 0.0348945 secs] [Times: user=0.05 sys=0.00, real=0.05 secs]
[Full GC [Tenured: 14335K->14335K(14336K), 0.0299813 secs] 19903K->19903K(19904K),
[Perm : 371K->371K(12288K)], 0.0300349 secs] [Times: user=0.01 sys=0.00, real=0.02 secs]
[Full GC [Tenured: 14335K->14335K(14336K), 0.0298178 secs] 19903K->19903K(19904K),
[Perm : 371K->371K(12288K)], 0.0298688 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
Exception in thread "main" java.lang.OutOfMemoryError: Java heap space[Full GC [Tenured:
14335K->14335K(14336K), 0.0294953 secs] 19903K->19903K(19904K),
[Perm : 371K->371K(12288K)], 0.0295474 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenured
: 14335K->14335K(14336K), 0.0287742 secs] 19903K->19903K(19904K),
[Perm : 371K->371K(12288K)], 0.0288239 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
[Full GC [Tenuredat GCTimeTest.main(GCTimeTest.java:16)
: 14335K->14335K(14336K), 0.0287102 secs] 19903K->19903K(19904K),
[Perm : 371K->371K(12288K)], 0.0287627 secs] [Times: user=0.03 sys=0.00, real=0.03 secs]
Heap
def new generation total 5568K, used 5567K [0x35c10000, 0x36210000, 0x36210000)
eden space 4992K, 100% used [0x35c10000, 0x360f0000, 0x360f0000)
from space 576K, 99% used [0x36180000, 0x3620ffe8, 0x36210000)
to space 576K, 0% used [0x360f0000, 0x360f0000, 0x36180000)
tenured generation total 14336K, used 14335K [0x36210000, 0x37010000, 0x37010000)
the space 14336K, 99% used [0x36210000, 0x3700ffd8, 0x37010000, 0x37010000)
compacting perm gen total 12288K, used 371K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706ce28, 0x3706d000, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
通过清单 2 和清单 3 对比，可以发现通过设置一个较大的年轻代预留新对象，设置合理的 Survivor 区并且提供 Survivor 区的使用率，可以将年轻对象保存在年轻代。一般来说，Survivor 区的空间不够，或者占用量达到 50%时，就会使对象进入年老代 (不管它的年龄有多大)。清单 4 创建了 3 个对象，分别分配一定的内存空间。
清单 4. 不同大小内存分配
public class PutInEden2 {
public static void main(String[] args){
byte[] b1,b2,b3;
b1=new byte[1024*512];//分配 0.5MB 堆空间
b2=new byte[1024*1024*4];//分配 4MB 堆空间
b3=new byte[1024*1024*4];
b3=null; //使 b3 可以被回收
b3=new byte[1024*1024*4];//分配 4MB 堆空间
}
}
使用参数-XX:+PrintGCDetails -Xmx1000M -Xms500M -Xmn100M -XX:SurvivorRatio=8 运行清单 4 所示代码，输出如清单 5 所示。
清单 5. 清单 4 运行输出
Heap
def new generation total 92160K, used 11878K [0x0f010000, 0x15410000, 0x15410000)
eden space 81920K, 2% used [0x0f010000, 0x0f1a9a20, 0x14010000)
from space 10240K, 99% used [0x14a10000, 0x1540fff8, 0x15410000)
to space 10240K, 0% used [0x14010000, 0x14010000, 0x14a10000)
tenured generation total 409600K, used 86434K [0x15410000, 0x2e410000, 0x4d810000)
the space 409600K, 21% used [0x15410000, 0x1a878b18, 0x1a878c00, 0x2e410000)
compacting perm gen total 12288K, used 2062K [0x4d810000, 0x4e410000, 0x51810000)
the space 12288K, 16% used [0x4d810000, 0x4da13b18, 0x4da13c00, 0x4e410000)
No shared spaces configured.
清单 5 输出的日志显示，年轻代分配了 8M，年老代也分配了 8M。我们可以尝试加上-XX:TargetSurvivorRatio=90 参数，这样可以提高 from 区的利用率，使 from 区使用到 90%时，再将对象送入年老代，运行清单 4 代码，输出如清单 6 所示。
清单 6. 修改运行参数后清单 4 输出
Heap
def new generation total 9216K, used 9215K [0x35c10000, 0x36610000, 0x36610000)
eden space 8192K, 100% used [0x35c10000, 0x36410000, 0x36410000)
from space 1024K, 99% used [0x36510000, 0x3660fc50, 0x36610000)
to space 1024K, 0% used [0x36410000, 0x36410000, 0x36510000)
tenured generation total 10240K, used 10239K [0x36610000, 0x37010000, 0x37010000)
the space 10240K, 99% used [0x36610000, 0x3700ff70, 0x37010000, 0x37010000)
compacting perm gen total 12288K, used 371K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706cd90, 0x3706ce00, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
如果将 SurvivorRatio 设置为 2，将 b1 对象预存在年轻代。输出如清单 7 所示。
清单 7. 再次修改运行参数后清单 4 输出
Heap
def new generation total 7680K, used 7679K [0x35c10000, 0x36610000, 0x36610000)
eden space 5120K, 100% used [0x35c10000, 0x36110000, 0x36110000)
from space 2560K, 99% used [0x36110000, 0x3638fff0, 0x36390000)
to space 2560K, 0% used [0x36390000, 0x36390000, 0x36610000)
tenured generation total 10240K, used 10239K [0x36610000, 0x37010000, 0x37010000)
the space 10240K, 99% used [0x36610000, 0x3700fff0, 0x37010000, 0x37010000)
compacting perm gen total 12288K, used 371K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706ce28, 0x3706d000, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
如何让大对象进入年老代
我们在大部分情况下都会选择将对象分配在年轻代。但是，对于占用内存较多的大对象而言，它的选择可能就不是这样的。因为大对象出现在年轻代很可能扰乱年轻代 GC，并破坏年轻代原有的对象结构。因为尝试在年轻代分配大对象，很可能导致空间不足，为了有足够的空间容纳大对象，JVM 不得不将年轻代中的年轻对象挪到年老代。因为大对象占用空间多，所以可能需要移动大量小的年轻对象进入年老代，这对 GC 相当不利。基于以上原因，可以将大对象直接分配到年老代，保持年轻代对象结构的完整性，这样可以提高 GC 的效率。如果一个大对象同时又是一个短命的对象，假设这种情况出现很频繁，那对于 GC 来说会是一场灾难。原本应该用于存放永久对象的年老代，被短命的对象塞满，这也意味着对堆空间进行了洗牌，扰乱了分代内存回收的基本思路。因此，在软件开发过程中，应该尽可能避免使用短命的大对象。可以使用参数-XX:PetenureSizeThreshold 设置大对象直接进入年老代的阈值。当对象的大小超过这个值时，将直接在年老代分配。参数-XX:PetenureSizeThreshold 只对串行收集器和年轻代并行收集器有效，并行回收收集器不识别这个参数。
清单 8. 创建一个大对象
public class BigObj2Old {
public static void main(String[] args){
byte[] b;
b = new byte[1024*1024];//分配一个 1MB 的对象
}
}
使用 JVM 参数-XX:+PrintGCDetails –Xmx20M –Xms20MB 运行，可以得到清单 9 所示日志输出。
清单 9. 清单 8 运行输出
Heap
def new generation total 6144K, used 1378K [0x35c10000, 0x362b0000, 0x362b0000)
eden space 5504K, 25% used [0x35c10000, 0x35d689e8, 0x36170000)
from space 640K, 0% used [0x36170000, 0x36170000, 0x36210000)
to space 640K, 0% used [0x36210000, 0x36210000, 0x362b0000)
tenured generation total 13696K, used 0K [0x362b0000, 0x37010000, 0x37010000)
the space 13696K, 0% used [0x362b0000, 0x362b0000, 0x362b0200, 0x37010000)
compacting perm gen total 12288K, used 374K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706dac8, 0x3706dc00, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
可以看到该对象被分配在了年轻代，占用了 25%的空间。如果需要将 1MB 以上的对象直接在年老代分配，设置-XX:PetenureSizeThreshold=1000000，程序运行后输出如清单 10 所示。
清单 10. 修改运行参数后清单 8 输出
Heap
def new generation total 6144K, used 354K [0x35c10000, 0x362b0000, 0x362b0000)
eden space 5504K, 6% used [0x35c10000, 0x35c689d8, 0x36170000)
from space 640K, 0% used [0x36170000, 0x36170000, 0x36210000)
to space 640K, 0% used [0x36210000, 0x36210000, 0x362b0000)
tenured generation total 13696K, used 1024K [0x362b0000, 0x37010000, 0x37010000)
the space 13696K, 7% used [0x362b0000, 0x363b0010, 0x363b0200, 0x37010000)
compacting perm gen total 12288K, used 374K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706dac8, 0x3706dc00, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
清单 10 里面可以看到当满 1MB 时进入到了年老代。
如何设置对象进入年老代的年龄
堆中的每一个对象都有自己的年龄。一般情况下，年轻对象存放在年轻代，年老对象存放在年老代。为了做到这点，虚拟机为每个对象都维护一个年龄。如果对象在 Eden 区，经过一次 GC 后依然存活，则被移动到 Survivor 区中，对象年龄加 1。以后，如果对象每经过一次 GC 依然存活，则年龄再加 1。当对象年龄达到阈值时，就移入年老代，成为老年对象。这个阈值的最大值可以通过参数-XX:MaxTenuringThreshold 来设置，默认值是 15。虽然-XX:MaxTenuringThreshold 的值可能是 15 或者更大，但这不意味着新对象非要达到这个年龄才能进入年老代。事实上，对象实际进入年老代的年龄是虚拟机在运行时根据内存使用情况动态计算的，这个参数指定的是阈值年龄的最大值。即，实际晋升年老代年龄等于动态计算所得的年龄与-XX:MaxTenuringThreshold 中较小的那个。清单 11 所示代码为 3 个对象申请了若干内存。
清单 11. 申请内存
public class MaxTenuringThreshold {
public static void main(String args[]){
byte[] b1,b2,b3;
b1 = new byte[1024*512];
b2 = new byte[1024*1024*2];
b3 = new byte[1024*1024*4];
b3 = null;
b3 = new byte[1024*1024*4];
}
}
参数设置为：-XX:+PrintGCDetails -Xmx20M -Xms20M -Xmn10M -XX:SurvivorRatio=2
运行清单 11 所示代码，输出如清单 12 所示。
清单 12. 清单 11 运行输出
[GC [DefNew: 2986K->690K(7680K), 0.0246816 secs] 2986K->2738K(17920K),
0.0247226 secs] [Times: user=0.00 sys=0.02, real=0.03 secs]
[GC [DefNew: 4786K->690K(7680K), 0.0016073 secs] 6834K->2738K(17920K),
0.0016436 secs] [Times: user=0.00 sys=0.00, real=0.00 secs]
Heap
def new generation total 7680K, used 4888K [0x35c10000, 0x36610000, 0x36610000)
eden space 5120K, 82% used [0x35c10000, 0x36029a18, 0x36110000)
from space 2560K, 26% used [0x36110000, 0x361bc950, 0x36390000)
to space 2560K, 0% used [0x36390000, 0x36390000, 0x36610000)
tenured generation total 10240K, used 2048K [0x36610000, 0x37010000, 0x37010000)
the space 10240K, 20% used [0x36610000, 0x36810010, 0x36810200, 0x37010000)
compacting perm gen total 12288K, used 374K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706db50, 0x3706dc00, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
更改参数为-XX:+PrintGCDetails -Xmx20M -Xms20M -Xmn10M -XX:SurvivorRatio=2 -XX:MaxTenuringThreshold=1，运行清单 11 所示代码，输出如清单 13 所示。
清单 13. 修改运行参数后清单 11 输出
[GC [DefNew: 2986K->690K(7680K), 0.0047778 secs] 2986K->2738K(17920K),
0.0048161 secs] [Times: user=0.00 sys=0.00, real=0.00 secs]
[GC [DefNew: 4888K->0K(7680K), 0.0016271 secs] 6936K->2738K(17920K),
0.0016630 secs] [Times: user=0.00 sys=0.00, real=0.00 secs]
Heap
def new generation total 7680K, used 4198K [0x35c10000, 0x36610000, 0x36610000)
eden space 5120K, 82% used [0x35c10000, 0x36029a18, 0x36110000)
from space 2560K, 0% used [0x36110000, 0x36110088, 0x36390000)
to space 2560K, 0% used [0x36390000, 0x36390000, 0x36610000)
tenured generation total 10240K, used 2738K [0x36610000, 0x37010000, 0x37010000)
the space 10240K, 26% used [0x36610000, 0x368bc890, 0x368bca00, 0x37010000)
compacting perm gen total 12288K, used 374K [0x37010000, 0x37c10000, 0x3b010000)
the space 12288K, 3% used [0x37010000, 0x3706db50, 0x3706dc00, 0x37c10000)
ro space 10240K, 51% used [0x3b010000, 0x3b543000, 0x3b543000, 0x3ba10000)
rw space 12288K, 55% used [0x3ba10000, 0x3c0ae4f8, 0x3c0ae600, 0x3c610000)
清单 13 所示，第一次运行时 b1 对象在程序结束后依然保存在年轻代。第二次运行前，我们减小了对象晋升年老代的年龄，设置为 1。即，所有经过一次 GC 的对象都可以直接进入年老代。程序运行后，可以发现 b1 对象已经被分配到年老代。如果希望对象尽可能长时间地停留在年轻代，可以设置一个较大的阈值。
稳定的 Java 堆 VS 动荡的 Java 堆
一般来说，稳定的堆大小对垃圾回收是有利的。获得一个稳定的堆大小的方法是使-Xms 和-Xmx 的大小一致，即最大堆和最小堆 (初始堆) 一样。如果这样设置，系统在运行时堆大小理论上是恒定的，稳定的堆空间可以减少 GC 的次数。因此，很多服务端应用都会将最大堆和最小堆设置为相同的数值。但是，一个不稳定的堆并非毫无用处。稳定的堆大小虽然可以减少 GC 次数，但同时也增加了每次 GC 的时间。让堆大小在一个区间中震荡，在系统不需要使用大内存时，压缩堆空间，使 GC 应对一个较小的堆，可以加快单次 GC 的速度。基于这样的考虑，JVM 还提供了两个参数用于压缩和扩展堆空间。
-XX:MinHeapFreeRatio 参数用来设置堆空间最小空闲比例，默认值是 40。当堆空间的空闲内存小于这个数值时，JVM 便会扩展堆空间。
-XX:MaxHeapFreeRatio 参数用来设置堆空间最大空闲比例，默认值是 70。当堆空间的空闲内存大于这个数值时，便会压缩堆空间，得到一个较小的堆。
当-Xmx 和-Xms 相等时，-XX:MinHeapFreeRatio 和-XX:MaxHeapFreeRatio 两个参数无效。
清单 14. 堆大小设置
import java.util.Vector;
public class HeapSize {
public static void main(String args[]) throws InterruptedException{
Vector v = new Vector();
while(true){
byte[] b = new byte[1024*1024];
v.add(b);
if(v.size() == 10){
v = new Vector();
}
Thread.sleep(1);
}
}
}
清单 14 所示代码是测试-XX:MinHeapFreeRatio 和-XX:MaxHeapFreeRatio 的作用，设置运行参数为-XX:+PrintGCDetails -Xms10M -Xmx40M -XX:MinHeapFreeRatio=40 -XX:MaxHeapFreeRatio=50 时，输出如清单 15 所示。
清单 15. 修改运行参数后清单 14 输出
[GC [DefNew: 2418K->178K(3072K), 0.0034827 secs] 2418K->2226K(9920K),
0.0035249 secs] [Times: user=0.00 sys=0.00, real=0.03 secs]
[GC [DefNew: 2312K->0K(3072K), 0.0028263 secs] 4360K->4274K(9920K),
0.0029905 secs] [Times: user=0.00 sys=0.00, real=0.03 secs]
[GC [DefNew: 2068K->0K(3072K), 0.0024363 secs] 6342K->6322K(9920K),
0.0024836 secs] [Times: user=0.00 sys=0.00, real=0.03 secs]
[GC [DefNew: 2061K->0K(3072K), 0.0017376 secs][Tenured: 8370K->8370K(8904K),
0.1392692 secs] 8384K->8370K(11976K), [Perm : 374K->374K(12288K)],
0.1411363 secs] [Times: user=0.00 sys=0.02, real=0.16 secs]
[GC [DefNew: 5138K->0K(6336K), 0.0038237 secs] 13508K->13490K(20288K),
0.0038632 secs] [Times: user=0.00 sys=0.00, real=0.03 secs]
改用参数：-XX:+PrintGCDetails -Xms40M -Xmx40M -XX:MinHeapFreeRatio=40 -XX:MaxHeapFreeRatio=50，运行输出如清单 16 所示。
清单 16. 再次修改运行参数后清单 14 输出
[GC [DefNew: 10678K->178K(12288K), 0.0019448 secs] 10678K->178K(39616K),
0.0019851 secs] [Times: user=0.00 sys=0.00, real=0.03 secs]
[GC [DefNew: 10751K->178K(12288K), 0.0010295 secs] 10751K->178K(39616K),
0.0010697 secs] [Times: user=0.00 sys=0.00, real=0.02 secs]
[GC [DefNew: 10493K->178K(12288K), 0.0008301 secs] 10493K->178K(39616K),
0.0008672 secs] [Times: user=0.00 sys=0.00, real=0.02 secs]
[GC [DefNew: 10467K->178K(12288K), 0.0008522 secs] 10467K->178K(39616K),
0.0008905 secs] [Times: user=0.00 sys=0.00, real=0.02 secs]
[GC [DefNew: 10450K->178K(12288K), 0.0008964 secs] 10450K->178K(39616K),
0.0009339 secs] [Times: user=0.00 sys=0.00, real=0.01 secs]
[GC [DefNew: 10439K->178K(12288K), 0.0009876 secs] 10439K->178K(39616K),
0.0010279 secs] [Times: user=0.00 sys=0.00, real=0.02 secs]
从清单 16 可以看出，此时堆空间的垃圾回收稳定在一个固定的范围。在一个稳定的堆中，堆空间大小始终不变，每次 GC 时，都要应对一个 40MB 的空间。因此，虽然 GC 次数减小了，但是单次 GC 速度不如一个震荡的堆。
增大吞吐量提升系统性能
吞吐量优先的方案将会尽可能减少系统执行垃圾回收的总时间，故可以考虑关注系统吞吐量的并行回收收集器。在拥有高性能的计算机上，进行吞吐量优先优化，可以使用参数：
java –Xmx3800m –Xms3800m –Xmn2G –Xss128k –XX:+UseParallelGC
–XX:ParallelGC-Threads=20 –XX:+UseParallelOldGC
–Xmx380m –Xms3800m：设置 Java 堆的最大值和初始值。一般情况下，为了避免堆内存的频繁震荡，导致系统性能下降，我们的做法是设置最大堆等于最小堆。假设这里把最小堆减少为最大堆的一半，即 1900m，那么 JVM 会尽可能在 1900MB 堆空间中运行，如果这样，发生 GC 的可能性就会比较高；
-Xss128k：减少线程栈的大小，这样可以使剩余的系统内存支持更多的线程；
-Xmn2g：设置年轻代区域大小为 2GB；
–XX:+UseParallelGC：年轻代使用并行垃圾回收收集器。这是一个关注吞吐量的收集器，可以尽可能地减少 GC 时间。
–XX:ParallelGC-Threads：设置用于垃圾回收的线程数，通常情况下，可以设置和 CPU 数量相等。但在 CPU 数量比较多的情况下，设置相对较小的数值也是合理的；
–XX:+UseParallelOldGC：设置年老代使用并行回收收集器。
尝试使用大的内存分页
CPU 是通过寻址来访问内存的。32 位 CPU 的寻址宽度是 0~0xFFFFFFFF ，计算后得到的大小是 4G，也就是说可支持的物理内存最大是 4G。但在实践过程中，碰到了这样的问题，程序需要使用 4G 内存，而可用物理内存小于 4G，导致程序不得不降低内存占用。为了解决此类问题，现代 CPU 引入了 MMU（Memory Management Unit 内存管理单元）。MMU 的核心**是利用虚拟地址替代物理地址，即 CPU 寻址时使用虚址，由 MMU 负责将虚址映射为物理地址。MMU 的引入，解决了对物理内存的限制，对程序来说，就像自己在使用 4G 内存一样。内存分页 (Paging) 是在使用 MMU 的基础上，提出的一种内存管理机制。它将虚拟地址和物理地址按固定大小（4K）分割成页 (page) 和页帧 (page frame)，并保证页与页帧的大小相同。这种机制，从数据结构上，保证了访问内存的高效，并使 OS 能支持非连续性的内存分配。在程序内存不够用时，还可以将不常用的物理内存页转移到其他存储设备上，比如磁盘，这就是大家耳熟能详的虚拟内存。
在 Solaris 系统中，JVM 可以支持 Large Page Size 的使用。使用大的内存分页可以增强 CPU 的内存寻址能力，从而提升系统的性能。
java –Xmx2506m –Xms2506m –Xmn1536m –Xss128k –XX:++UseParallelGC
–XX:ParallelGCThreads=20 –XX:+UseParallelOldGC –XX:+LargePageSizeInBytes=256m
–XX:+LargePageSizeInBytes：设置大页的大小。
过大的内存分页会导致 JVM 在计算 Heap 内部分区（perm, new, old）内存占用比例时，会出现超出正常值的划分，最坏情况下某个区会多占用一个页的大小。
使用非占有的垃圾回收器
为降低应用软件的垃圾回收时的停顿，首先考虑的是使用关注系统停顿的 CMS 回收器，其次，为了减少 Full GC 次数，应尽可能将对象预留在年轻代，因为年轻代 Minor GC 的成本远远小于年老代的 Full GC。
java –Xmx3550m –Xms3550m –Xmn2g –Xss128k –XX:ParallelGCThreads=20
–XX:+UseConcMarkSweepGC –XX:+UseParNewGC –XX:+SurvivorRatio=8 –XX:TargetSurvivorRatio=90
–XX:MaxTenuringThreshold=31
–XX:ParallelGCThreads=20：设置 20 个线程进行垃圾回收；
–XX:+UseParNewGC：年轻代使用并行回收器；
–XX:+UseConcMarkSweepGC：年老代使用 CMS 收集器降低停顿；
–XX:+SurvivorRatio：设置 Eden 区和 Survivor 区的比例为 8:1。稍大的 Survivor 空间可以提高在年轻代回收生命周期较短的对象的可能性，如果 Survivor 不够大，一些短命的对象可能直接进入年老代，这对系统来说是不利的。
–XX:TargetSurvivorRatio=90：设置 Survivor 区的可使用率。这里设置为 90%，则允许 90%的 Survivor 空间被使用。默认值是 50%。故该设置提高了 Survivor 区的使用率。当存放的对象超过这个百分比，则对象会向年老代压缩。因此，这个选项更有助于将对象留在年轻代。
–XX:MaxTenuringThreshold：设置年轻对象晋升到年老代的年龄。默认值是 15 次，即对象经过 15 次 Minor GC 依然存活，则进入年老代。这里设置为 31，目的是让对象尽可能地保存在年轻代区域。
结束语
通过本文的学习，读者了解了如何将新对象预留在年轻代、如何让大对象进入年老代、如何设置对象进入年老代的年龄、稳定的 Java 堆 VS 动荡的 Java 堆、增大吞吐量提升系统性能、尝试使用大的内存分页、使用非占有的垃圾回收器等主题，通过实例及对应输出解释的形式让读者对于 JVM 优化有一个初步认识。如其他文章相同的观点，没有哪一条优化是固定不变的，读者需要自己判断、实践后才能找到正确的道路。
关于easymock的典型使用方式，在easymock的官网文档中，有非常详尽的讲解，文档地址为 http://easymock.org/EasyMock3_0_Documentation.html，文档的开头一部分内容都是easymock中最基本的使用介绍，虽然是英文，但是非常容易看懂，适用新学者入门。
这里只罗列一些简单的常用功能，依然以前面教程中使用到的测试案例为例：
public class UserServiceImplTest extends Assert {
@Test
public void testQuery() {
User expectedUser = new User();
user.set.;
UserDao userDao = Easymock.createMock(UserDao.class);
Easymock.expect(userDao.getById("1001")).andReturn(expectedUser);
Easymock.replay(userDao);
UserServiceImpl service = new UserServiceImpl();
service.setUserDao(userDao);
user user = service.query("1001");
assertNotNull(user);
assertEquals(); //veify return user
Easymock.verify(userDao);
}
}
这段简短的代码中包含以下easymock的功能：
1. 创建mock对象
UserDao userDao = Easymock.createMock(UserDao.class);
2. 记录mock对象期望的行为
Easymock.expect(userDao.getById("1001")).andReturn(expectedUser);
这里记录了mock对象的行为：getById()方法被调用，调用次数为1(easymock之中如果没有明确指出调用次数，默认为1),参数为"1001"，expectedUser将作为返回值。
3. 进入replay阶段
Easymock.replay(userDao);
4. 对mock对象执行验证
Easymock.verify(userDao);
对上面上面的代码稍加改动以展示easymock的其他基本功能
1. 指定期望的调用次数
Easymock.expect(userDao.getById("1001")).andReturn(expectedUser).times(3);
2. 指定抛出期望的异常
Easymock.expect(userDao.getById("1001")).andThrow(new RuntimeException("no user exist"));
3. 记录void 方法的行为
Easymock.expect(userDao.getById("1001")) 这样的用法只能使用与mock对象的有返回值的方法，如果mock对象的方法是void，则需要使用expectLastCall()：
userDao.someVoidMethod();
Easymock.expectLastCall();
和Easymock.expect(***)一样，同样支持指定调用次数，抛出异常等：
Easymock.expectLastCall().times(3);
Easymock.expectLastCall().andThrow(new RuntimeException("some error"));
4. 灵活的参数匹配
Easymock.expect(userDao.getById(Easymock.isA(String.class))).andReturn(expectedUser);
类似的还有anyInt()，anyObject()， isNull() ， same(), startsWith()等诸多实现。
在最近的项目中看到同事使用到了Disruptor，以前在ifeve上看到过关于Disruptor的文章，但是没有深入研究，现在项目中用到了，就借这个机会对这个并发编程框架进行深入学习。项目中使用到的是disruptor-2.10.4，所以下面分析到的Disruptor的代码是这个版本的。
并发编程网介绍Disruptor的文章是disruptor1.0版本，所以有一些术语在2.0版本上已经没有了或者被替代了。
Disruptor术语
github上Disruptor的wiki对Disruptor中的术语进行了解释，在看Disruptor的过程中，对于几个其他的类，觉得有必要与这些术语放到一起，就加进来了。
RingBuffer 经常被看作Disruptor最主要的组件，然而从3.0开始RingBuffer仅仅负责存储和更新在Disruptor中流通的数据。对一些特殊的使用场景能够被用户(使用其他数据结构)完全替代。
Sequence Disruptor使用Sequence来表示一个特殊组件处理的序号。和Disruptor一样，每个消费者(EventProcessor)都维持着一个Sequence。大部分的并发代码依赖这些Sequence值的运转，因此Sequence支持多种当前AtomicLong类的特性。事实上，这两者之间唯一的区别是Sequence包含额外的功能来阻止Sequence和其他值之间的共享。
Sequencer 这是Disruptor真正的核心。实现了这个接口的两种生产者（单生产者和多生产者）均实现了所有的并发算法，为了在生产者和消费者之间进行准确快速的数据传递。
SequenceBarrier 由Sequencer生成，并且包含了已经发布的Sequence的引用，这些的Sequence源于Sequencer和一些独立的消费者的Sequence。它包含了决定是否有供消费者来消费的Event的逻辑。
WaitStrategy：它决定了一个消费者将如何等待生产者将Event置入Disruptor。
Event 从生产者到消费者过程中所处理的数据单元。Disruptor中没有代码表示Event，因为它完全是由用户定义的。
EventProcessor 主要的事件循环，用于处理Disruptor中的Event，并且拥有消费者的Sequence。它有一个实现类是BatchEventProcessor，包含了event loop有效的实现，并且将回调到一个EventHandler接口的实现对象。
EventHandler 由用户实现并且代表了Disruptor中的一个消费者的接口。
Producer 由用户实现，它调用RingBuffer来插入事件(Event)，在Disruptor中没有相应的实现代码，由用户实现。
WorkProcessor 确保每个sequence只被一个processor消费，在同一个WorkPool中的处理多个WorkProcessor不会消费同样的sequence。
WorkerPool 一个WorkProcessor池，其中WorkProcessor将消费Sequence，所以任务可以在实现WorkHandler接口的worker吃间移交
LifecycleAware 当BatchEventProcessor启动和停止时，于实现这个接口用于接收通知。
Disruptor印象
初看Disruptor，给人的印象就是RingBuffer是其核心，生产者向RingBuffer中写入元素，消费者从RingBuffer中消费元素，如下图：
这就是Disruptor最简单的模型。其中的RingBuffer被组织成要给环形队列，但它与我们在常常使用的队列又不一样，这个队列大小固定，且每个元素槽都以一个整数进行编号，RingBuffer中只有一个游标维护着一个指向下一个可用位置的序号，生产者每次向RingBuffer中写入一个元素时都需要向RingBuffer申请一个可写入的序列号，如果此时RingBuffer中有可用节点，RingBuffer就向生产者返回这个可用节点的序号，如果没有，那么就等待。同样消费者消费的元素序号也必须是生产者已经写入了的元素序号。
那么Disruptor是如何实现这些逻辑的呢？先来看一个Disruptor的使用示例
Disruptor使用示例
不适用Disruptor的dsl，直接使用Disruptor中的类来完成。
[java] view plain copy
//RingBuffer中存储的单元
public class IntEvent {
private int value = -1;
public int getValue() {
return value;
}
public void setValue(int value) {
this.value = value;
}
public String toString() {
return String.valueOf(value);
}
public static EventFactory<IntEvent> INT_ENEVT_FACTORY = new EventFactory<IntEvent>() {
public IntEvent newInstance() {
return new IntEvent();
}
};
}
//生产者
public class IntEventProducer implements WorkHandler {
private int seq = 0;
public void onEvent(IntEvent event) throws Exception {
System.out.println("produced " + seq);
event.setValue(++seq);
}
}
//消费者
public class IntEventProcessor implements WorkHandler {
public void onEvent(IntEvent event) throws Exception {
System.out.println(event.getValue());
event.setValue(1);
}
}
public class DisruptorTest {
public static void main(String[] args) throws InterruptedException {
//创建一个RingBuffer对象
RingBuffer<IntEvent> ringBuffer = new RingBuffer<IntEvent>(IntEvent.INT_ENEVT_FACTORY,
new SingleThreadedClaimStrategy(16),
new SleepingWaitStrategy());
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
IntEventProducer[] producers = new IntEventProducer[1];
for (int i = 0; i < producers.length; i++) {
producers[i] = new IntEventProducer();
}
WorkerPool<IntEvent> crawler = new WorkerPool<IntEvent>(ringBuffer,
sequenceBarrier,
new IntEventExceptionHandler(),
producers);
SequenceBarrier sb = ringBuffer.newBarrier(crawler.getWorkerSequences());
IntEventProcessor[] processors = new IntEventProcessor[1];
for (int i = 0; i < processors.length; i++) {
processors[i] = new IntEventProcessor();
}
WorkerPool<IntEvent> applier = new WorkerPool<IntEvent>(ringBuffer,sb,
new IntEventExceptionHandler(),
processors);
List<Sequence> gatingSequences = new ArrayList<Sequence>();
for(Sequence s : crawler.getWorkerSequences()) {
gatingSequences.add(s);
}
for(Sequence s : applier.getWorkerSequences()) {
gatingSequences.add(s);
}
ringBuffer.setGatingSequences(gatingSequences.toArray(new Sequence[gatingSequences.size()]));
ThreadPoolExecutor executor = new ThreadPoolExecutor(7,7,10,TimeUnit.MINUTES,new LinkedBlockingQueue(5));
crawler.start(executor);
applier.start(executor);
while (true) {
Thread.sleep(1000);
long lastSeq = ringBuffer.next();
ringBuffer.publish(lastSeq);
}
}
}
class IntEventExceptionHandler implements ExceptionHandler {
public void handleEventException(Throwable ex, long sequence, Object event) {}
public void handleOnStartException(Throwable ex) {}
public void handleOnShutdownException(Throwable ex) {}
}
在上面的代码中IntEvent类就是术语中的Event，IntEventProducer对应Producer，IntEventProcessor对应着EventProcessor，也就是消费者，但是IntEventProcessor类并不是实现的IntEventProcessor接口，这个下面会分析到。
下面从main方法开始分析。
首先是RingBuffer的创建，RingBuffer ringBuffer = new RingBuffer(IntEvent.INT_ENEVT_FACTORY,new SingleThreadedClaimStrategy(16), new SleepingWaitStrategy());RingBuffer的构造方法的第一个参数是实现了EventFactory接口的类，主要作用是创建IntEvent对象，在创建RingBuffer对象时，第二个参数是ClaimStrategy，生产者通过ClaimStrategy 来申请下一个可用节点，第三个参数是WaitStrategy的实现类，它定义了消费者的等待策略。
下面一段代码是初始化生产者的过程
[java] view plain copy
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
IntEventProducer[] producers = new IntEventProducer[1];
for (int i = 0; i < producers.length; i++) {
producers[i] = new IntEventProducer();
}
WorkerPool crawler = new WorkerPool(ringBuffer,
sequenceBarrier,new IntEventExceptionHandler(), producers);
ringBuffer.newBarrier()返回的是一个SequenceBarrier对象，Barrier顾名思义就是障碍的意思，这个障碍阻止了生产者越过其所能到达的ringbuffer中的位置。RingBuffer在执行过程中会维持一个游标cursor，所以生产者从RingBuffer中获取到的的游标必须小于等于这个cursor。
然后就是消费者的初始化：
[java] view plain copy
SequenceBarrier sb = ringBuffer.newBarrier(crawler.getWorkerSequences());
IntEventProcessor[] processors = new IntEventProcessor[1];
for (int i = 0; i < processors.length; i++) {
processors[i] = new IntEventProcessor();
}
WorkerPool applier = new WorkerPool(ringBuffer,sb,new IntEventExceptionHandler(),processors);
消费者初始化也需要设置一个SequenceBarrier对象，这个SequenceBarrier对象指明了消费者可以消费的元素序号，如果消费者的游标大于这个序号，那么消费者必须以WaitStrategy定义的策略等待。
生产者和消费者创建完成后下一步是设置RingBuffer的一个变量gatingSequences，gatingSequences的作用是防止生产者覆盖还未被消费者消费的元素，假设一个RingBuffer的大小为8，消费者消费速度较慢，那么RingBuffer可能是满的，当生产者向RingBuffer申请下一个可用序号时，还未被消费者消费的序号就不能被覆盖，所以RingBuffer就不能给生产者返回可用序号，此时消费者线程就进入等待，在这种情况下RingBuffer会检查当前申请的序号是否大于gatingSequences中的最小序号，如果当前申请的序号大于最小序号，那么生产者就等待。代码如下：
[java] view plain copy
List gatingSequences = new ArrayList();
for(Sequence s : crawler.getWorkerSequences()) {
gatingSequences.add(s);
}
for(Sequence s : applier.getWorkerSequences()) {
gatingSequences.add(s);
} ringBuffer.setGatingSequences(gatingSequences.toArray(new Sequence[gatingSequences.size()]));
这里之所以加入了生产者的序号，是因为在多生产者的情况下，一个生产者不能覆盖另一个生产者已经申请的序号。
然后下面一段的代码就是启动Disruptor线程来执行生产者和消费者线程。
[java] view plain copy
ThreadPoolExecutor executor = new ThreadPoolExecutor(7, 7,10,
TimeUnit.MINUTES, new LinkedBlockingQueue(2));
crawler.start(executor);
applier.start(executor);
最后要做的是生产者不断的申请可用序号，提交准备好的元素序号，即：
[java] view plain copy
while (true) {
Thread.sleep(1000);
long lastSeq = ringBuffer.next();
ringBuffer.publish(lastSeq);
}
这个过程一般成为两阶段提交，第一个阶段调用RingBuffer.next()方法得到下一个可用节点，此时会调用生产者类的OnEvent方法，即IntEventProducer.onEvent()，第二个阶段是发布已经生产完成的元素序号lastSeq，提醒消费者，可以消费这个序号了。
小结
本文简要介绍了Disruptor的一些术语，解释了一个Disruptor示例程序执行过程的各个要点，从宏观上了解了Disruptor的执行流程，在下一篇文章中接着分析Disruptor的内部执行原理。
概述
Redis-Sentinel是Redis官方推荐的高可用性(HA)解决方案，当用Redis做Master-slave的高可用方案时，假如master宕机了，Redis本身(包括它的很多客户端)都没有实现自动进行主备切换，而Redis-sentinel本身也是一个独立运行的进程，它能监控多个master-slave集群，发现master宕机后能进行自懂切换。
它的主要功能有以下几点
不时地监控redis是否按照预期良好地运行;
如果发现某个redis节点运行出现状况，能够通知另外一个进程(例如它的客户端);
能够进行自动切换。当一个master节点不可用时，能够选举出master的多个slave(如果有超过一个slave的话)中的一个来作为新的master,其它的slave节点会将它所追随的master的地址改为被提升为master的slave的新地址。
无failover时的配置纠正
即使当前没有failover正在进行，sentinel依然会使用当前配置去设置监控的master。特别是：
根据最新配置确认为slaves的节点却声称自己是master(参考上文例子中被网络隔离后的的redis3)，这时它们会被重新配置为当前master的slave。
如果slaves连接了一个错误的master，将会被改正过来，连接到正确的master。
Slave选举与优先级
当一个sentinel准备好了要进行failover，并且收到了其他sentinel的授权，那么就需要选举出一个合适的slave来做为新的master。
slave的选举主要会评估slave的以下几个方面：
与master断开连接的次数
Slave的优先级
数据复制的下标(用来评估slave当前拥有多少master的数据)
进程ID
如果一个slave与master失去联系超过10次，并且每次都超过了配置的最大失联时间(down-after-milliseconds option)，并且，如果sentinel在进行failover时发现slave失联，那么这个slave就会被sentinel认为不适合用来做新master的。
更严格的定义是，如果一个slave持续断开连接的时间超过
(down-after-milliseconds * 10) + milliseconds_since_master_is_in_SDOWN_state
就会被认为失去选举资格。
符合上述条件的slave才会被列入master候选人列表，并根据以下顺序来进行排序：
sentinel首先会根据slaves的优先级来进行排序，优先级越小排名越靠前（？）。
如果优先级相同，则查看复制的下标，哪个从master接收的复制数据多，哪个就靠前。
如果优先级和下标都相同，就选择进程ID较小的那个。
一个redis无论是master还是slave，都必须在配置中指定一个slave优先级。要注意到master也是有可能通过failover变成slave的。
如果一个redis的slave优先级配置为0，那么它将永远不会被选为master。但是它依然会从master哪里复制数据。
Sentinel和Redis身份验证
当一个master配置为需要密码才能连接时，客户端和slave在连接时都需要提供密码。
master通过requirepass设置自身的密码，不提供密码无法连接到这个master。
slave通过masterauth来设置访问master时的密码。
但是当使用了sentinel时，由于一个master可能会变成一个slave，一个slave也可能会变成master，所以需要同时设置上述两个配置项。
Sentinel API
Sentinel默认运行在26379端口上，sentinel支持redis协议，所以可以使用redis-cli客户端或者其他可用的客户端来与sentinel通信。
有两种方式能够与sentinel通信：
一种是直接使用客户端向它发消息
另外一种是使用发布/订阅模式来订阅sentinel事件，比如说failover，或者某个redis实例运行出错，等等。
Sentinel命令
sentinel支持的合法命令如下：
PING sentinel回复PONG.
SENTINEL masters 显示被监控的所有master以及它们的状态.
SENTINEL master 显示指定master的信息和状态；
SENTINEL slaves 显示指定master的所有slave以及它们的状态；
SENTINEL get-master-addr-by-name 返回指定master的ip和端口，如果正在进行failover或者failover已经完成，将会显示被提升为master的slave的ip和端口。
SENTINEL reset 重置名字匹配该正则表达式的所有的master的状态信息，清楚其之前的状态信息，以及slaves信息。
SENTINEL failover 强制sentinel执行failover，并且不需要得到其他sentinel的同意。但是failover后会将最新的配置发送给其他sentinel。
动态修改Sentinel配置
从redis2.8.4开始，sentinel提供了一组API用来添加，删除，修改master的配置。
需要注意的是，如果你通过API修改了一个sentinel的配置，sentinel不会把修改的配置告诉其他sentinel。你需要自己手动地对多个sentinel发送修改配置的命令。
以下是一些修改sentinel配置的命令：
SENTINEL MONITOR 这个命令告诉sentinel去监听一个新的master
SENTINEL REMOVE 命令sentinel放弃对某个master的监听
SENTINEL SET 这个命令很像Redis的CONFIG SET命令，用来改变指定master的配置。支持多个。例如以下实例：
SENTINEL SET objects-cache-master down-after-milliseconds 1000
只要是配置文件中存在的配置项，都可以用SENTINEL SET命令来设置。这个还可以用来设置master的属性，比如说quorum(票数)，而不需要先删除master，再重新添加master。例如：
SENTINEL SET objects-cache-master quorum 5
增加或删除Sentinel
由于有sentinel自动发现机制，所以添加一个sentinel到你的集群中非常容易，你所需要做的只是监控到某个Master上，然后新添加的sentinel就能获得其他sentinel的信息以及masterd所有的slave。
如果你需要添加多个sentinel，建议你一个接着一个添加，这样可以预防网络隔离带来的问题。你可以每个30秒添加一个sentinel。最后你可以用SENTINEL MASTER mastername来检查一下是否所有的sentinel都已经监控到了master。
删除一个sentinel显得有点复杂：因为sentinel永远不会删除一个已经存在过的sentinel，即使它已经与组织失去联系很久了。
要想删除一个sentinel，应该遵循如下步骤：
停止所要删除的sentinel
发送一个SENTINEL RESET * 命令给所有其它的sentinel实例，如果你想要重置指定master上面的sentinel，只需要把*号改为特定的名字，注意，需要一个接一个发，每次发送的间隔不低于30秒。
检查一下所有的sentinels是否都有一致的当前sentinel数。使用SENTINEL MASTER mastername 来查询。
删除旧master或者不可达slave
sentinel永远会记录好一个Master的slaves，即使slave已经与组织失联好久了。这是很有用的，因为sentinel集群必须有能力把一个恢复可用的slave进行重新配置。
并且，failover后，失效的master将会被标记为新master的一个slave，这样的话，当它变得可用时，就会从新master上复制数据。
然后，有时候你想要永久地删除掉一个slave(有可能它曾经是个master)，你只需要发送一个SENTINEL RESET master命令给所有的sentinels，它们将会更新列表里能够正确地复制master数据的slave。
发布/订阅
客户端可以向一个sentinel发送订阅某个频道的事件的命令，当有特定的事件发生时，sentinel会通知所有订阅的客户端。需要注意的是客户端只能订阅，不能发布。
订阅频道的名字与事件的名字一致。例如，频道名为sdown 将会发布所有与SDOWN相关的消息给订阅者。
如果想要订阅所有消息，只需简单地使用PSUBSCRIBE *
以下是所有你可以收到的消息的消息格式，如果你订阅了所有消息的话。第一个单词是频道的名字，其它是数据的格式。
注意：以下的instance details的格式是：
@
如果这个redis实例是一个master，那么@之后的消息就不会显示。
+reset-master <instance details> -- 当master被重置时.
+slave <instance details> -- 当检测到一个slave并添加进slave列表时.
+failover-state-reconf-slaves <instance details> -- Failover状态变为reconf-slaves状态时
+failover-detected <instance details> -- 当failover发生时
+slave-reconf-sent <instance details> -- sentinel发送SLAVEOF命令把它重新配置时
+slave-reconf-inprog <instance details> -- slave被重新配置为另外一个master的slave，但数据复制还未发生时。
+slave-reconf-done <instance details> -- slave被重新配置为另外一个master的slave并且数据复制已经与master同步时。
-dup-sentinel <instance details> -- 删除指定master上的冗余sentinel时 (当一个sentinel重新启动时，可能会发生这个事件).
+sentinel <instance details> -- 当master增加了一个sentinel时。
+sdown <instance details> -- 进入SDOWN状态时;
-sdown <instance details> -- 离开SDOWN状态时。
+odown <instance details> -- 进入ODOWN状态时。
-odown <instance details> -- 离开ODOWN状态时。
+new-epoch <instance details> -- 当前配置版本被更新时。
+try-failover <instance details> -- 达到failover条件，正等待其他sentinel的选举。
+elected-leader <instance details> -- 被选举为去执行failover的时候。
+failover-state-select-slave <instance details> -- 开始要选择一个slave当选新master时。
no-good-slave <instance details> -- 没有合适的slave来担当新master
selected-slave <instance details> -- 找到了一个适合的slave来担当新master
failover-state-send-slaveof-noone <instance details> -- 当把选择为新master的slave的身份进行切换的时候。
failover-end-for-timeout <instance details> -- failover由于超时而失败时。
failover-end <instance details> -- failover成功完成时。
switch-master <master name> <oldip> <oldport> <newip> <newport> -- 当master的地址发生变化时。通常这是客户端最感兴趣的消息了。
+tilt -- 进入Tilt模式。
-tilt -- 退出Tilt模式。
TILT 模式
redis sentinel非常依赖系统时间，例如它会使用系统时间来判断一个PING回复用了多久的时间。
然而，假如系统时间被修改了，或者是系统十分繁忙，或者是进程堵塞了，sentinel可能会出现运行不正常的情况。
当系统的稳定性下降时，TILT模式是sentinel可以进入的一种的保护模式。当进入TILT模式时，sentinel会继续监控工作，但是它不会有任何其他动作，它也不会去回应is-master-down-by-addr这样的命令了，因为它在TILT模式下，检测失效节点的能力已经变得让人不可信任了。
如果系统恢复正常，持续30秒钟，sentinel就会退出TITL模式。
-BUSY状态
注意：该功能还未实现。
当一个脚本的运行时间超过配置的运行时间时，sentinel会返回一个-BUSY 错误信号。如果这件事发生在触发一个failover之前，sentinel将会发送一个SCRIPT KILL命令，如果script是只读的话，就能成功执行。
红包架构
1.预资源下载
这些页面需要用到图片，视频或H5页面等资源。在活动期间，参与用户多，对资源的请求量很大，如果都通过实时在线访问，服务器的网络带宽会面临巨大压力，基本无法支撑；另外，资源的尺寸比较大，下载到手机需要较长时间，用户体验也会很差。因此,我们采用预先下载的方式, 在活动开始前几天把资源推送给客户端,客户端在需要使用时直接从本地 加载。
2.摇／拆红包
除夕的摇一摇子系统是专门为活动定制的,按时间轴进行各项活动,这里边最重要、同时也是请求量最大的是摇红包。从需求上看,系统需要完成两个事:用户可以通过摇一摇抢到红包,红包金额可以入到用户的支付账户。在除夕,系统需要在很短时间内将几十亿个红包发放下去,对性能和可用性要求很高。考虑到涉及资金的业务逻辑比较复杂,还有很多数据库事务处理,耗时会比较长,于是我们将抢红包(信息流)和红包的账务逻辑(业务流和资金流)异步化。将前一部分处理流程尽可能设计得轻量,让用户可以很快抢到红包,然后再异步完成剩下的账务逻辑。
在微信后台系统中,一般情况下客户端发起的请求都是通过接入服务
转发给具体的业务服务处理的,会产生 RPC 调用。但对于摇一摇请求,我 们将摇一摇逻辑直接嵌入接入服务中,接入服务可以直接处理摇一摇请求, 派发红包。
按一般的系统实现,用户看到的红包在系统中是数据库中的数据记录,
抢红包就是找出可用的红包记录,将该记录标识为属于某个用户。在这种
实现里,数据库是系统的瓶颈和主要成本开销。我们在这一过程完全不使 用数据库,可以达到几个数量级的性能提升,同时可靠性有了更好的保障。
支付系统将所有需要下发的红包生成红包票据文件 ;
一）.将红包票据文件拆分后放到每一个接入服务实例中;
二）.接收到客户端发起摇一摇请求后，接入服务里的摇一摇逻辑拿出一个红包票据，在本地生成一个跟用户绑定的加密票据，下发给客户端；
三）.客户端拿加密票据到后台拆红包，后台的红包简化服务通过本地计算即可验证红包，完成抢红包过程。
3.异步化
用户抢到红包后不会同步进行后续的账务处理，请求会被放入红包异步队列，再通过异步队列转给微信支付后台,由微信支付后台完成后续业务逻辑。
二 . 大规模集群中保证数据一致性
1 . 服务性能
为提升各个服务模块的处理性能,我们通过压测和 Profiler 分析, 发现了不少性能瓶颈点,做了大量优化。
2 . 业务支撑能力
支持更加复杂的业务场景,并在客户端和服务器都加入了很多可以后 期灵活调整的预埋能力,以更好地服务产品运营。
不断提升系统可用性是我们一直努力的目标,以下 5 点很好地提高了 系统的可用性。
一、背景
当前， 几乎所有的手机应用都不会是单机运行在机器中，而是配合一个或者多个后台服务共同完成所有操作。举个例子，一个微博客户端，在用户进行看帖、发帖、顶踩等操作时，都会将请求发送给微博的后台服务器进行处理。这种交互方式就是典型的C/S架构，交互图大概是介样子的：
然而实际应用场景中，并不会只有一个后台服务，而是多个服务分而治之。比如上面例子中的微博客户端，可能会与后台的帖子服务器、点赞服务器、图片服务器（不要在意命名这些细节）等多个服务器产生交互请求。于是，这个时候的交互图变成了：
上面都还只是提到了内部服务，那如果再加上一些必要的外联系统呢（比如广告服务、新闻服务、内容推荐服务、搜索服务等第三方服务）？可以看到交互图有些丧心病狂了：
对于客户端来说，无论是内部服务还是外联服务，都是一些独立的外部server，客户端无法干涉也不应该干涉。那么在客户端的开发或测试过程中，应该如何让这些server“配合”做测试和联调呢？最简单粗暴的方式就是直接连接外部系统。但是这种方式有很大的缺陷：我们无法控制这些外部系统，只能覆盖到最基本的场景。同时这种方式也受到了很大的限制，万一这些外部系统还未开发完成或者暂时无法提供服务，那我们的联调、测试工作将被block住。所以，最可靠的方式就是自己构造或者模拟一些这样的server，然后再随意控制server的响应以覆盖所有联调和测试场景。这便是mock平台的作用。
二、现有mock平台的局限和新解决方案
当前业界也存在许多优秀的mock平台或工具，比如GoogleMock, EasyMock, MockServer等，但是这些mock平台工具多少存在着一些局限性：
三、整体架构和设计思路
名词解释
MCS：Mock Control Server，即Mock平台的管理主控点，专用于维护、管理Mock节点，提供API配置、管理Mock数据。
MSN：Mock Service Node，mock服务节点，顾名思义，就是实际接收mock请求和响应请求的mock节点。MSN与MCS之间也通过http请求进行交互和通信。
设计思路
四、缓存模式
本文所描述的mock平台设计初衷是完成一个高性能的分布式mock平台。分布式体现在MSN散步于不同机器，避免单一节点造成数据冲突和集中访问带来的性能瓶颈。但是从之前架构图和交互流程可以看出，MSN在处理mock请求时需要向MCS查询对应的响应数据，如果有一个应用持续高频地发送大量请求到MSN，也会造成MSN大量请求MCS，再加上众多MSN的心跳请求，MCS势必会成为整个方案的瓶颈所在。为了解决这个问题，让mock平台成为一个支持性能测试的真正意义上的高性能分布式mock平台，我们引入了mock数据缓存模式的概念。
缓存模式的实现比较简单：在配置mock数据的时候，增加缓存标识，如果用户设置缓存模式开启，则MSN在第一次查询到该mock数据时便将数据缓存到MSN内存中（需要控制缓存数据的数量和大小，避免将机器内存打满）。也就是说，MSN在处理mock请求的时候，实际上是先查找本地缓存，没有找到才会去查询MCS。因此，如果需要进行性能测试，只需将数据配置成缓存模式即可。这样不但省去了MSN请求MCS这一个步骤，还将读取MCS数据库变成了读取内存，性能和响应速度直接提升几个数量级。
缓存数据的清除就更加简单了。首先，MSN提供接口触发缓存数据的清除。其次，当释放、重新分配MSN时会进行环境的初始化，这时候不仅会清除MSN中的缓存，还会清除MCS中给该MSN配置的mock数据。
五、二期扩展功能
一期实现暂时只是一个比较简单的mock平台，不过已经五脏俱全，可以满足基本的mock需求了。
二期规划主要增加3个功能点：随机响应、其它协议支持以及代理录制。
随机响应
随机响应指通过某种约定方式配置mock数据，让MSN收到请求时可以随机生成响应返回给应用。举个例子，假设应用请求的真实返回是json格式的：
{
"retCode": 0,
"retMsgs": "success",
"retData": 100
}
可见retData是一个整数，如果正常情况下接口返回的retData范围是1 ~ 65535，那么就可以设置retData为一个1 ~ 65535之间的正整数，MSN在收到请求时便可以为retData选取一个1 ~ 65535之间的随机数，这样便完成了一个随机响应。其它数据类型如float、double、string等亦类似，不再赘叙。
随机响应功能的适用场景有：稳定性测试、健壮性测试和性能测试。
其它协议支持
有些时候，应用与后台服务的交互并不是采用http协议，而是采用其它一些协议如tcp socket, soap, rpc, thrift等。又或者不仅仅使用http而是采用更加安全的https协议。
为了提升mock平台的通用性，二期增加多种协议支持，为产品线定制开发其它各种协议的MSN节点。
代理录制
从上面的架构和详细设计中可以看出，用户在配置好mock后还需要修改应用的配置才能将请求指向MSN。如果应用是一个web server或者二进制程序，一般情况下只需要修改配置后重启即可。但如果应用是一个手机app，还要修改配置后重新编译打包、安装部署，整个过程较为耗时。为了减少这些麻烦，二期规划将MSN改造成代理服务器，用户只需要为手机配置好代理网络指向MSN即可使用mock服务，无需修改任何应用配置。这就相当于MSN截获了应用的所有网络请求，通过解析请求协议类型和请求url、路径和参数，向MCS查询到相应的mock结果后返回给应用，这个过程对应用来说是完全透明的。
既然能够截获所有请求，MSN还可以做的一件事就是开启完全代理模式，将请求转发给实际后台server，并记录所有请求、响应数据。用户后续可以一键将这些请求响应数据转化为mock数据配置，这些配置代表了真实返回，可以用作性能测试和场景测试，当数据量大的时候，可以省去很多数据构造工作。
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft
Open source projects and samples from Microsoft.
Google
Google ❤️ Open Source for everyone.
Alibaba
Alibaba Open Source for everyone
D3
Data-Driven Documents codes.
Tencent
China tencent open source team.