分布式服务器框架

通信框架,序列化与反序列化,协议栈,服务路由,集群容错,服务调用,服务注册中心,服务发布和引用,服务灰度发布,参数传递,服务多版本,流量控制,服务降级,服务优先级调度,服务治理,分布式消息跟踪,可靠性设计,

分布式服务架构设计

抽象三层
RPC层,Filter Chain层,Service层
功能上还有服务治理中心,服务注册中心

功能特性

服务订阅发布:配置化发布和引用,服务自动化发现机制,服务在线注册和去注册
服务路由:默认提供随机路由、轮询、基于权重的路由策略等,黏滞连接,路由定制
集群容错:failover,failback,failfast
服务调用:同步调用,异步调用,并行调用
多协议:私有协议、公有协议
序列化方式:二进制类序列化,文本类序列化
统一配置:本地静态配置,基于配置中心的动态配置

通读一遍书籍,写一下第一遍概览
几个章节

远古科技

黎明之前:

从用石头砸开坚果开始、告别茹毛饮血、从山洞到茅屋、虱子与衣服、最早的武器、善于说话者胜

文明曙光:

存储太阳能、驯化水稻、陶器的出现、美索不达米亚的水渠、轮子与帆船、从1到10、开始记录一切、星辰的轨迹

古代科技

农耕文明:

青铜与铁、解决粮食问题、纺织,瓷器与玻璃、城市的胜利

文明复兴:

古希腊人的贡献、纸张与文明、从雕版印刷到活字印刷、阿拉伯文明的黄金时代、大学的诞生、漫长的中世纪、美第奇与佛罗伦萨的穹顶、日心说:冲破教会束缚

近代科技

科学启蒙:

科学方法论,近代医学的诞生,开启大航海时代,神说、让牛顿说吧!,炼金术士还是化学家

工业革命:

月光社与工业革命,蒸汽船与火车,机器思维的胜利,永动机不存在,细胞学说与进化论,电的发现与存储,直流电与交流电,通信以光速进行

新工业:

石油:工业的血液,化学工业助力农业,给轮子上加上内燃机,飞上篮天,杀伤力跃迁,洪堡与教育改革

现代科技

原子时代:

突破物理学危机,了不起的原子能,雷达的本质,制药业突飞猛进

信息时代:

新数学和新方法论,从算盘到机械计算机,电子计算机的诞生,摩尔定律的动力,互联网,移动通信之路,太空竞赛,从豌豆杂交开始的基因技术

未来世界:

癌症的预测性检测,基因编辑,可控核聚变,“新生产关系”区块链,量子通信实现数据安全,

历史总是反反复复, 但科技一直向前,科技是唯一可叠加的进步。近代科技之前,人类发展缓慢,在掌握了科学方法之后,科技水平飞速发展,而且之后会越来越快。
虽然人类科技已经进化了很多次,但人如果不接受教育,还保留着原始人基因的特性,很多方法与原始人很相近,要想能接近社会的平均水平,就要学习人类的历史,不去重复之前的错误,用最新的研究成果,才能起点就很高。
每一个历史节点都需要细细拆解,明白里面的原理,反观自己。
我一直想知道怎么才能了解历史,怎么去查资料,现在还是吃别人的食物,要自己去找食物,现在还不知道,吴军老师是怎么查资料的。有个思路就是查他的附录,里面有很多引用,可以深入学习。
看完再来重新回顾,突然感觉有了一种更高的视野,原来一切就是这样。
掌握思维方法最重要,机械思维,不确定思维,控制论、信息轮、系统论。掌握的更先进。

反观很多人连机械思维还没拥有,还用原始大脑去思考,那必然会落后于时代。

背景

扩展单个电脑单个程序的过程调用,到多台电脑上

优点

简单,干净的语义,高效,通用型。

设计系统需要的问题,

面对机器和通信失败时的精确语义
包含地址参数在缺少共享地址时的语义
将远程调用集成到现有系统中
绑定,调用者怎么决定哪个地址和标识的被调用者
在调用者和被调用者之间传输和控制数据的合适的协议
如何在开发的网络中提供完整性和开发性

目标

最主要的目的是简化分布式计算,以前这个工作通常需要经验丰富的专家,很难做。
希望像本地访问,移除不必要的困难,只留下构建分布式系统的基本困难:时序,组件的独立故障以及独立执行环境的共存。
次要目的是,RPC沟通是高效的
另外,RPC 包的语义尽可能强大,不失简单和高效。
最后的目标是建立安全的连接。
共享内存有2个潜在的问题
首先,远程地址的表示是否可以集成到我们的编程语言(可能还有底层的机器架构)中,而不会产生过度的剧变;
第二,是否可以实现可接受的效率。

结构

user ,user-stub,RPCRuntime,server-stub,server

总结

通读了论文全部,然后再细读,论文包含了,NDREW D. BIRRELL and BRUCE JAY NELSON 对整个系统设计时的思考,能清楚的知道怎么做,以及为什么这么做,设计

看了一遍《分布式服务框架-原理与实践》这本书,大致浏览了一遍有了一些初略的印象。
RPC应用在分布式部署服务,互相需要调用的情况下,原来是在同一个进程间互相调用,互相通信,但不同进程,多机器就不能用原来的进程间调用的方法,但又想跟原来进程间一样简单,这样RPC就诞生了。
RPC分为几个部分,序列化与反序列化,网络框架,协议栈。简单来说就是先序列化,然后网络传输,然后再反序列化。
那这个服务框架的好坏可以再细分了序列化的好坏,网络模型的好坏,网络协议的好坏。
一个问题的研究,需要将这个问题拆分成小的问题,再每个问题深入研究,最后合成一个整体考虑。我们把RPC进行拆分,一个表象的问题,最终会拆分成基础的问题,网络问题,与网络传输序列化问题 。

最近再复习各种计算机相关的,有些知识比如计算机原理,网络、操作系统,我也知道这些知识很重要是根本的东西,但工作上不是那么继续,时间有限,这些东西总会往后拖,拖到现在工作来6年了,这些知识还是没有体系。
最近再想想是不是要考研,一个方面是自己工作到了一个瓶颈,想想怎么能向上走,另外一方面,正好趁着考试,把这些科班需要的知识体系补全,这样我的计算机方面的知识体系就全面。即使考试没成功,知识体系也补全了,能有一个目标去努力一些不那么急切需要的东西很不容易。
但我有考虑,这种为了考试的学习,是不是能真正掌握知识,还是只是精通考试,这个很重要,我不是为了考试,但为了达到目标,可能会偏离初心,所以学习过程很重要,过程大于目标。如果还像之前学习辅修那样补习,即使课程学习完成,还是收获很小,那次学习得到了结果,但没有过程,对于未来起到的作用很小,所以这次考试我要以正确的方式去学习,那怎样才能算正确的过程学习?
分析之前的学习经验,在大学,学习了很多全面的知识体系,但那个时候不知道为什么学习这些,只知道有这些,然后不断做题训练,在离开书本后一无所知,虽然知识学了很多,但不知道怎么使用,学习知识很枯燥,导致了很快失去了兴趣,后面只是机械的学习,这是之前导致失败的原因。与之不同的工作后的学习,大部分是应用,快速的使用别人的东西,马上上手,很快应用,知识碎片化,表象化。很多东西都会,但遇到需要分析的细节就是一脸懵逼,没有理解细节,只是随机试错,效率很低,关键问题不知道该如何下手。
分析这两种不同的学习经验,发现走了2个极端,理论与实践,两个互相混合,才能理解,现实原因决定理论发展,理论发展加快现实的进度。懂得系统的知识可以让你站在巨人的肩膀上进步,起点比较高,不用重新趟别人错过的坑,加快进步的速度,但知道知识会让人迷茫,脱离实际,理论不能应用在实际情况上就毫无意义。
那实际学习中要如何结合呢,光靠课本是达不到这点的,要我们自己去找食物。要先找到构建这门学科的人,找到他是如何建立基础概念的,过程是怎么样的,不能单单看他构建的知识,要看他为什么这么构建,怎么构建,中间经历过什么错误,从历史中学习,但这些材料可能不好查找。这么做的目的就是将概念与实际现实结合起来,从实际中学习概念是如何建立,学习整个知识体系的构建过程,不单单是知道有这些知识,能讲出来概念的由来,这样你就能很好的理论联系实际。然后是学习概念的细节内涵,如果以考试为目的,那就需要大量的训练,记忆,但不能经历错误的过程,建立知识体系,理论结合实际是我们的初心,大量训练是为了熟练,考试结果也只是结果。如果只达到了结果,跟原来的辅修的结果可能差不多,要为了更长远的利益达成。
准备复习考研的方法,第一遍,以理解概念为主,找历史记录,找知识体系构建者的建立概念的过程为主,以理解为主,小量训练为辅,构建知识体系。
第二遍,开始大量训练,找到知识体系漏洞,记录错题,修正错误理解。
第三遍,以错题,与考试大纲为主,构建体系与查缺补漏。

一个线程单独的开放寻址探测的hashmap,本地存储变量,每个线程不共享变量,就没有了并发问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
private final int threadLocalHashCode = nextHashCode();

/**
* The next hash code to be given out. Updated atomically. Starts at
* zero.
*/
private static AtomicInteger nextHashCode =
new AtomicInteger();

/**
* The difference between successively generated hash codes - turns
* implicit sequential thread-local IDs into near-optimally spread
* multiplicative hash values for power-of-two-sized tables.
*/
private static final int HASH_INCREMENT = 0x61c88647;

/**
* Returns the next hash code.
*/
private static int nextHashCode() {
return nextHashCode.getAndAdd(HASH_INCREMENT);
}

ThreadLocal里面有这个几个内部变量,threadLocalHashCode是这个线程本地的hashCode值,这个值是取nextHashCode这个原子变量,初始值是0,每次递增HASH_INCREMENT。就是说每个线程从创建依次递增0x61c88647,这个数据很有来头,是32位有符号数字的黄金比例,为了让hash值能更好的分散,不出现冲突。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public T get() {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
return setInitialValue();
}
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}

每个线程thread里有一个变量是threadLocals,通过当前threadlocal对象为key,获取存储的ThreadLocalMap,里面的value,就是我们存储的变量。

1
2
3
4
5
6
7
8
9
10
11
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}
void createMap(Thread t, T firstValue) {
t.threadLocals = new ThreadLocalMap(this, firstValue);
}

Set方法跟get原理差不多,主要是这个ThreadLocalMap

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
static class ThreadLocalMap {

/**
* The entries in this hash map extend WeakReference, using
* its main ref field as the key (which is always a
* ThreadLocal object). Note that null keys (i.e. entry.get()
* == null) mean that the key is no longer referenced, so the
* entry can be expunged from table. Such entries are referred to
* as "stale entries" in the code that follows.
*/
static class Entry extends WeakReference<ThreadLocal<?>> {
/** The value associated with this ThreadLocal. */
Object value;

Entry(ThreadLocal<?> k, Object v) {
super(k);
value = v;
}
}

/**
* The initial capacity -- MUST be a power of two.
*/
private static final int INITIAL_CAPACITY = 16;

/**
* The table, resized as necessary.
* table.length MUST always be a power of two.
*/
private Entry[] table;

可以看出来这个ThreadLocal的内部类ThreadLocalMap,里面有一个Entry是弱引用ThreadLocal,Entry强引用value。ThreadLocalMap是使用数组作为底层的存储结构。我当时还想为什么没有继承hashmap,原来是hash算法不一致,hashmap使用拉链法解决冲突,而这里使用的开放寻址法,这种算法在大量冲突的时候,效率比较低,但我们前面提到的hashcode算法,有效的让hash值不特别冲突,而且本地变量应该与hashmap这样的数据结构不一样,没有那么大的存储量级,所以这样效率更高。

总结:基于数组存储的,开放寻址法的哈希散列。每个散列值存储在线程本地,通过ThreadLocal代理去线程内获取变量。
这次先看总的结构,还有些里面的清理方法没有完全看懂,下次继续。

日本版 观看感受

我只看到了28岁的日本版,抛开对个人的评论, 说一个总体的影响,日本也是有一个多个民族融合的问题,美国占领,建军事基地。俄罗斯的领土问题,与中国人口的融合。
日本整体老龄化严重,有些僻远的地方没有年轻人,只有老年人,人数越来越少。
大城市人口越来越多,人的生存压力很大,工作家庭很难两全。
社会整体是下降的,年轻人找不到上升的阶梯,但还是有一些人把握住了机会,能靠自己做成一些事情。
与英国的社会有差异,但也有共性问题。

南非版 观看感受

南非是民族问题很严重的一个地方,白人与黑人,开始还有种族隔离问题,暴力流血随处可见,在曼德拉上台后有好转。
白人的生活很惬意,黑人大部分还在生存边缘挣扎。
战争,艾滋病影响着人们的生存,这些是日本和英国没有的。情况在逐渐好转,但治安问题还是大家所担心的。没有工作,也是黑人所担心的,没有好的教育,没有挣钱能力才是贫穷的根源。
28岁时有3个孩子去世了,1个是因为医疗落后,2个是因为艾滋病。
白人与黑人,种族不同,没有本质的差别,主要差别在于是否接受过良好的教育,几个有钱的黑人家庭,给孩子良好的教育,未来也都很好。

题目

给定一个非空字符串 s 和一个包含非空单词列表的字典 wordDict,判定 s 是否可以被空格拆分为一个或多个在字典中出现的单词。
说明:

  • 拆分时可以重复使用字典中的单词。
  • 你可以假设字典中没有重复的单词。
    示例 1:
    输入: s = “leetcode”, wordDict = [“leet”, “code”]
    输出: true
    解释: 返回 true 因为 “leetcode” 可以被拆分成 “leet code”。
    示例 2:
    输入: s = “applepenapple”, wordDict = [“apple”, “pen”]
    输出: true
    解释: 返回 true 因为 “applepenapple” 可以被拆分成 “apple pen apple”。
    注意你可以重复使用字典中的单词。
    示例 3:
    输入: s = “catsandog”, wordDict = [“cats”, “dog”, “sand”, “and”, “cat”]
    输出: false

解题思路

先将wordDict根据首字母分类
然后先从s的第一个字母找起,有一个比s长一位的布尔数组,递归判断如果 最后一个是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
class Solution {
boolean[] exist;
Map<Character,List<Integer>> map;
List<String> wordDict;
char[] sArray;
public boolean wordBreak(String s, List<String> wordDict) {
exist = new boolean[s.length()+1];
map = new HashMap<Character,List<Integer>>();
this.wordDict = wordDict;
for(int i = 0 ; i< wordDict.size() ; i++){
Character first =wordDict.get(i).charAt(0);
if(!map.containsKey(first)){
List list = new ArrayList<Integer>();
list.add(i);
map.put(first,list);
}else{
List<Integer> oldList = map.get(first);
oldList.add(i);
}
}
sArray = s.toCharArray();
innerCheck(0);
return exist[s.length()];
}
public void innerCheck(int i){
if(i == sArray.length){
return;
}
char n = sArray[i];
List<Integer> nList = map.get(n);
if(nList == null){
return;
}
outer:
for(Integer num: nList){
String match = wordDict.get(num);
char[] matchArray = match.toCharArray();
for(int j = 0 ;j < matchArray.length ; j++){
if(i+j >= sArray.length || sArray[i+j] != matchArray[j] ){
continue outer;
}
}
if(exist[i+matchArray.length]){
return;
}
exist[i+matchArray.length] = true;
innerCheck(i+matchArray.length);
}
}
}

动态规划还有点难掌握,需要多练习,现在的感觉是主要是能利用之前的状态。

题目

给定一个无重复元素的数组 candidates 和一个目标数 target ,找出 candidates 中所有可以使数字和为 target 的组合。
candidates 中的数字可以无限制重复被选取。
说明:

  • 所有数字(包括 target)都是正整数。
  • 解集不能包含重复的组合。
    示例 1:
    输入: candidates = [2,3,6,7], target = 7,
    所求解集为:
    [
    [7],
    [2,2,3]
    ]
    示例 2:
    输入: candidates = [2,3,5], target = 8,
    所求解集为:
    [
    [2,2,2,2],
    [2,3,3],
    [3,5]
    ]

解题思路

关键是结束条件,每次遍历target-candidates[I],传入下一个递归,如何等于0 ,结束,如果小于最小值也结束
深度遍历,可以重复,如果返回,则删除最后一位添加的,继续下一个的计算

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
class Solution {
List list;
List item;
public List<List<Integer>> combinationSum(int[] candidates, int target) {
list = new ArrayList<>();
item = new ArrayList<Integer>();
if(candidates == null || candidates.length == 0){
return list;
}
Arrays.sort(candidates);
backTracing(0,candidates,target);
return list;

}
private void backTracing(int index ,int[] candidates , int target){
if(target == 0){
list.add(new ArrayList<>(item));
return;
}
if(target < candidates[0]){
return;
}

for(int i = index ; i< candidates.length ; i++){
int targetDelete = target - candidates[i];
item.add(candidates[i]);
backTracing(i, candidates, targetDelete);
item.remove(item.size()-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
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:
* {@code corePoolSize < 0}
* {@code keepAliveTime < 0}
* {@code maximumPoolSize <= 0}
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}

构造器有6个参数,corePoolSize是核心线程数,maximumPoolSize是最大线程数,keepAliveTime是空闲线程保持时间,unit保持时间单位,workQueue是排队队列,threadFactory自定义创建线程工厂,handler拒绝策略。

1
2
3
4
5
6
7
8
9
10
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;

ctl是 线程池允许状态和线程数组合,前3位是状态,后29位是线程数,用一个原子变量表示。

主要的方法,执行线程方法

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
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}

execute方法,提交runnable接口的对象
主要3种处理方式,
1.线程池的运行线程小于corePoolSize,运行addWorker添加新的线程
2.如果大于corePoolSize,同时线程池还在运行,就添加到工作队列里,添加成功后还要检查状态。如果线程池不允许了,就拒绝任务。如果运行线程等于0了,就新建执行任务线程。
3.如果线程池添加队列失败,就添加线程直到最大线程数,去执行任务。如果无法新建线程就拒绝任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());

if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

addWorker方法,用于添加线程,先是一个不断自旋去工作线程数,修改ctl。通过core 参数,判断是与corePoolSize 还是maximumPoolSize去判断大小,一个是是否达到核心线程数,一个是否达到最大线程数。
修改状态完成后,通过break label退出无限循环。
后面是新建worker内部类,将新建的worker添加到workers里,一个hashset,里面包含所有的只要执行的worker,worker只有获取了mainLock锁才能访问。
最后启动线程执行start方法,会运行worker里的run方法。

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
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}

run方法执行runWorker方法,将自身传入。
getTask从阻塞队列里获取任务,后面就是执行task里的run方法执行。

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
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?

for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}

int wc = workerCountOf(c);

// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

不断自旋判断,如果线程池停止或者队列为空就减少工作线程数。
获取线程有2种情况,timed是allowCoreThreadTimeOut 和工作线程是否大于核心线程来决定。
2种情况,1种所有线程都记录空闲时间,1种是只有超过核心线程记录空闲时间。
这两种情况使用poll方法,如果队列为空就阻塞keepAliveTime时间,就返回。
如果不是这两种情况,就直接take如果阻塞就直接报错,然后重新循环。