ThreadPoolExecutor线程池 实现原理

类结构

首先要先了解一下类结构，如下图：

• Executor

1
2
3

public interface Executor {
    void execute(Runnable command);
}

只有一个接口，传入一个Runnable对象，线程池就会帮你执行这个指令。

• ExecutorService

1
2
3
4
5
6
7
8
9
10
11

public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
 
    .......省略........
}

这个接口是执行器服务接口，声明了关于执行器的许多管理方法。

• AbstractExecutorService

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

public abstract class AbstractExecutorService implements ExecutorService {
  
  .......省略........
    
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
  
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
  
 .......省略........
}

这个抽象类实现了ExecutorService接口中的大部分方法，不过大部分的实现都依赖于Executor接口声明的execute方法，而这里并没有实现这个关键的方法，而是把这个方法的实现交给了子类，也就是java.util.concurrent.ThreadPoolExecutor来实现了。

LinkedHashMap 源码阅读

概要

• 类继承关系

1
2
3

public class LinkedHashMap<K,V>
       extends HashMap<K,V>
       implements Map<K,V>

LinkedHashMap继承自HashMap

• 核心成员变量

jdk 7:(下文都是该环境)

1
2
3
4
5
6

/**
 * The head of the doubly linked list.
 */
private transient Entry<K,V> header;   //链表头结点
final boolean accessOrder;   //按元素插入顺序(默认)或元素最近访问顺序(LRU)排列

jdk 8:

1
2
3
4
5

transient LinkedHashMap.Entry<K,V> head;   //链表头结点
transient LinkedHashMap.Entry<K,V> tail;  //链表尾节点
final boolean accessOrder;

jdk 8中新增了一个链表尾节点

• 特点

一般来说，如果需要使用的Map中的key无序，选择HashMap；如果要求key有序，则选择TreeMap。
但是选择TreeMap就会有性能问题，因为TreeMap的get操作的时间复杂度是O(log(n))的，相比于HashMap的O(1)还是差不少的，LinkedHashMap的出现就是为了平衡这些因素，使得

能够以O(1)时间复杂度增加查找元素，又能够保证key的有序性

此外，LinkedHashMap提供了两种key的顺序：

访问顺序（access order）：非常实用，可以使用这种顺序实现LRU（Least Recently Used）缓存

插入顺序（insertion orde）：同一key的多次插入，并不会影响其顺序

多connect-factory时配置exchange， queue需使用declared-by

背景：

项目中需要监听多个rabbitmq服务器或者发消息到多个rabbitmq服务器，项目启动报错。结果如下：

1
2
3
4
5
6
7

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - cannot redeclare exchange 'lamia.live.start' in vhost '/' with different type, durable, internal or autodelete value, class-id=40, method-id=10)
at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:478)
at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:315)
at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:144)
at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:91)
at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:550)
... 1 more

问题原因：从错误来看，大概意思是重复声明了“exchange”。研究了下spring-rabbit中“exchange”的配置。

基本信息：

Broker：简单来说就是消息队列服务器实体，可以把一个rabbitmq server当作一个broker。

vhost虚拟主机：一个broker里可以开设多个vhost，用作不同用户的权限分离，也可以把他看作明名空间。vhost之间相互完全隔离，不同Vhost之间无法共享Exchange和Queue。(如果不指定vhost,默认是”/“)

Exchange：Exchange是属于Vhost的。同一个vhost不能有重复的Exchange名称(这个就是错误原因)。

Channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

什么是 Binlog

MySQL Server有四种类型的日志——Error Log、General Query Log、Binary Log 和 Slow Query Log。

　　第一个是错误日志，记录 mysqld (mysql daemon)的一些错误。第二个是一般查询日志，记录 mysqld 正在做的事情，比如客户端的连接和断开、来自客户端每条 Sql Statement 记录信息；如果你想知道客户端到底传了什么给服务端，这个日志就非常管用了，不过它非常影响性能。第四个是慢查询日志，记录一些查询比较慢的 SQL 语句——这种日志非常常用，主要是给开发者调优用的。

　　第三种就是 Binlog 了，binlog是mysql的二进制日志，包含了一些事件，这些事件描述了数据库的改动，如建表、数据改动等，也包括一些潜在改动，比如 DELETE FROM ran WHERE bing = luan，然而一条数据都没被删掉的这种情况。除非使用 Row-based logging，否则会包含所有改动数据的 SQL Statement。

注意：binlog主要有statement-based、row-based logging和mixed logging 三种格式，row-based的记录中不包括潜在更新记录。

　　 Binlog 有两个重要的用途——复制和恢复。比如主从表的复制，和备份恢复。

注：mysqld是MySQL服务器的守护进程。

binlog格式

可以指定三种binary log的格式(启动时指定)：

1
2
3

--binlog-format=STATEMENT
--binlog-format=ROW
--binlog-format=MIXED

• statement-based logging： 基于SQL语句，Mysql5.6默认，某些语句和函数如UUID, LOAD DATA INFILE等在复制过程可能导致数据不一致甚至出错。
• row-based logging：基于行，记录影响table中每一行的事务，很安全。所以一条语句可能会对应0-N个事件，记录很详细，数据同步的支持比STATEMENT方式要好。但是binlog会比其他两种模式大很多，在一些大表中清除大量数据时在binlog中会生成很多条语句，可能导致从库延迟变大。
• mixed logging：使用statement-based logging作为默认，但是日志模式可能会在某些情况下自动切换到row-based logging。

转载：CopyOnWriteArrayList与JMM

1 什么是CopyOnWriteArrayList

CopyOnWriteArrayList相当于线程安全的ArrayList，是一个可变数组。它具有如下特性：

• 是线程安全的
• 写操作会复制整个基础数组，因此写操作开销很大
• 适用于如下情况：数组大小较小，并且读操作比写操作多很多的情形

2 CopyOnWriteArrayList的设计原理与JMM

下面分析CopyOnWriteArrayList的设计原理，结合JMM的基础知识，分析CopyOnWriteArrayList是如何保证线程安全的。

首先看用来实际保存数据的数组：

1
2

/** The array, accessed only via getArray/setArray. */
private transient volatile Object[] array;

可以看到array数组前面使用了volatile变量来修饰。volatile主要用来解决内存可见性问题。关于volatile的详细实现原理可以参考《深入理解java内存模型.pdf》以及Java并发编程：volatile关键字解析-博客园-海子。

2.1 CopyOnWriteArrayList的读方法

读方法比较简单，直接从array中获取对应索引的值。

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

/**
* {@inheritDoc}
*
* @throws IndexOutOfBoundsException {@inheritDoc}
*/
public E get(int index) {
return get(getArray(), index);
}
@SuppressWarnings("unchecked")
private E get(Object[] a, int index) {
return (E) a[index];
}
/**
* Gets the array. Non-private so as to also be accessible
* from CopyOnWriteArraySet class.
*/
final Object[] getArray() {
return array;
}

2.2 CopyOnWriteArrayList的写方法

• set方法
  ​

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

/**
 * Replaces the element at the specified position in this list with the
 * specified element.
 *
 * @throws IndexOutOfBoundsException {@inheritDoc}
 */
public E set(int index, E element) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        E oldValue = get(elements, index);
        if (oldValue != element) {
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len);
            newElements[index] = element;
            setArray(newElements);
        } else {
            // Not quite a no-op; ensures volatile write semantics
            setArray(elements);
        }
        return oldValue;
    } finally {
        lock.unlock();
    }
}
/**
 * Sets the array.
 */
final void setArray(Object[] a) {
    array = a;
}

set方法的功能是将对应索引的元素置为一个新值。执行流程：
（1）加锁
（2）获取对应索引已有的值
（3）比较已有的值和新值，如果不相等，转4，否则转5
（4）创建新的数组，复制原数组的元素，并将对应索引置为新值。然后将新数组赋给array（setArray）
（5）setArray:将array赋给array

这里有一个比较奇怪的点，为什么已有的值和新值相等的时候，还要执行setArray呢？本质上setArray也没有做什么事情。

这段代码混合使用了锁以及volatile。锁的用法比较容易理解，它在使用同一个锁的不同线程之间保证内存顺序性，代码结尾的释放锁的操作提供了本线程和其他欲获取相同的锁的线程之间的happens-before语义。但是CopyOnWriteArrayList类中其他代码，不一定会使用到这把锁，因此，前面所述的锁带来的内存模型含义对这部分代码执行是不适用的。其他没用到这把锁的代码，读写是volatile读和volatile写（因为array前面使用volatile关键字修饰）。由volatile来保证happens-before语义。

volatile的特性及原理

volatile 变量自身具有下列特性:
（1）可见性。对一个volatile 变量的读,总是能看到(任意线程)对这个volatile变量最后的写入。
（2）原子性:对任意单个volatile 变量的读/写具有原子性,但类似于volatile++这种复合操作不具有原子性。

volatile 写和锁的释放有相同的内存语义。

为了实现 volatile 的内存语义,编译器在生成字节码时,会在指令序列中插入内存屏障来禁止特定类型的处理器重排序。

这里调用setArray的原因是，确保set方法对array执行的永远是volatile写。这就和其他对array执行volatile读的线程之间建立了happens-before语义。非常重要的一点：volatile读/写语义针对的是读写操作，而不是使用volatile修饰的变量本身。这样说更直白一点：在一个volatile写操作之前的对其他非volatile变量的写，happens-before于同一个volatile变量读操作之后的对其他变量的读。这句话比较绕，看下面一个例子就比较易懂了。

1
2
3
4
5
6
7
8
9
10
11

// initial conditions
int nonVolatileField = 0;
CopyOnWriteArrayList<String> list = /* a single String */
// Thread 1
nonVolatileField = 1;                 // (1)
list.set(0, "x");                     // (2)
// Thread 2
String s = list.get(0);               // (3)
if (s == "x") {
    int localVar = nonVolatileField;  // (4)
}

现在假设原始数组中无元素“x”，这样(2)成功设置了元素”x”，(3)处可以成功获取到元素”x”。这种情况下，(4)一定会读取到(1)处设置的值1.因为(2)处的volatile写以及在此之前的任何写操作都happens-before(3)处的读以及之后的所有读。

但是，假设一开始数组中就有了元素”x”，如果else不调用setArray，那么(2)处的写就不是volatile写，(4)处的读就不一定能读到(1)处设置的值！

很显然我们不想让内存可见性依赖于list中已有的值，为了确保任何情况下的内存可见性，set方法必须永远都是一个volatile写，这就是为何要在else代码块中调用setArray的原因。