java并发与多线程

Executor接口

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

    Executor接口中之定义了一个方法execute（Runnable command），该方法接收一个Runable实例，它用来执行一个任务，任务即一个实现了Runnable接口的类。 在Java 5之后，任务分两类：一类是实现了Runnable接口的类，一类是实现了Callable接口的类。两者都可以被ExecutorService执行，但是Runnable任务没有返回值，而Callable任务有返回值。并且Callable的call()方法只能通过ExecutorService的submit(Callable<T> task) 方法来执行，并且返回一个 <T>Future<T>，是表示任务等待完成的Future。    Callable接口类似于Runnable，两者都是为那些其实例可能被另一个线程执行的类设计的。但是 Runnable 不会返回结果，并且无法抛出经过检查的异常而Callable又返回结果，而且当获取返回结果时可能会抛出异常。Callable中的call()方法类似Runnable的run()方法，区别同样是有返回值，后者没有。    当将一个Callable的对象传递给ExecutorService的submit方法，则该call方法自动在一个线程上执行，并且会返回执行结果Future对象。同样，将Runnable的对象传递给ExecutorService的submit方法，则该run方法自动在一个线程上执行，并且会返回执行结果Future对象，但是在该Future对象上调用get方法，将返回null。

ExecutorService接口

public interface ExecutorService extends Executor {
    void shutdown();
    List shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
    Future submit(Callable task);
    Future submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    List<Future> invokeAll(Collection<? extends Callable> tasks) throws InterruptedException;
    List<Future> invokeAll(Collection<? extends Callable> tasks, long timeout, TimeUnit unit) throws InterruptedException;
    T invokeAny(Collection<? extends Callable> tasks) throws InterruptedException, ExecutionException;
    T invokeAny(Collection<? extends Callable> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

    一共12个方法，其中一部分是和执行器生命周期相关的方法，而另一部分则是以各种方式提交要执行的任务的方法。像submit()就是提交任务的一个方法，在实现中做了适配的工作，无论参数是Runnable还是Callable，执行器都会正确执行。当然，这实际上用到的是前文提过的RunnableFuture的实现类FutureTask。
    ExecutorService接口继承自Executor接口，它提供了更丰富的实现多线程的方法，比如，ExecutorService提供了关闭自己的方法，以及可为跟踪一个或多个异步任务执行状况而生成 Future 的方法。可以调用ExecutorService的shutdown()方法来平滑地关闭 ExecutorService，调用该方法后，将导致ExecutorService停止接受任何新的任务且等待已经提交的任务执行完成(已经提交的任务会分两类：一类是已经在执行的，另一类是还没有开始执行的)，当所有已经提交的任务执行完毕后将会关闭ExecutorService。因此我们一般用该接口来实现和管理多线程。
    ExecutorService的生命周期包括三种状态：运行、关闭、终止。创建后便进入运行状态，当调用了shutdown()方法时，便进入关闭状态，此时意味着ExecutorService不再接受新的任务，但它还在执行已经提交了的任务，当素有已经提交了的任务执行完后，便到达终止状态。如果不调用shutdown()方法，ExecutorService会一直处在运行状态，不断接收新的任务，执行新的任务，服务器端一般不需要关闭它，保持一直运行即可。
    那为什么要使用ExecutorService呢？
    a. 每次new Thread新建对象性能差。
    b. 线程缺乏统一管理，可能无限制新建线程，相互之间竞争，及可能占用过多系统资源导致死机或oom。
    c. 缺乏更多功能，如定时执行、定期执行、线程中断。
    相比new Thread，Java提供的四种线程池的好处在于： 
    a. 重用存在的线程，减少对象创建、消亡的开销，性能佳。
    b. 可有效控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。
    c. 提供定时执行、定期执行、单线程、并发数控制等功能。

ThreadPoolExecutor

    自定义线程池，可以用ThreadPoolExecutor类创建，它有多个构造方法来创建线程池，用该类很容易实现自定义的线程池
    //创建等待队列 
    BlockingQueue<Runnable> bqueue = new ArrayBlockingQueue<Runnable>(20); 
    //创建线程池，池中保存的线程数为3，允许的最大线程数为5 
    ThreadPoolExecutor pool = new ThreadPoolExecutor(3,5,50,TimeUnit.MILLISECONDS,bqueue);

AbstractExecutorService

    这个类是一个抽象类，规范了所有 工厂（Executors）构造的实例，也就是所有工厂构建的ExecutorService实例必须继承自AbstractExecutorService。这个类是ExecutorService的一个抽象实现。其中，提交任务的各类方法已经给出了十分完整的实现。之所以抽象，是因为和执行器本身生命周期相关的方法，在此类中并未给出任何实现，需要子类扩展完善。

ScheduleExecutorService & ScheduledThreadPoolExecutor

    用于执行定时任务

Executors

    Executors提供了一系列工厂方法用于创先线程池，返回的线程池都实现了ExecutorService接口。这些方法返回的ExecutorService对象最终都是由ThreadPoolExecutor实现的，根据不同的需求以不同的参数配置，或经过其它类包装。

    // 创建固定数目线程的线程池。
    public static ExecutorService newFixedThreadPool(int nThreads)
    // 创建一个可缓存的线程池，调用execute将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线 程并添加到池中。终止并从缓存中移除那些已 有 60 秒钟未被使用的线程。
    public static ExecutorService newCachedThreadPool()
    // 创建一个单线程化的Executor。
    public static ExecutorService newSingleThreadExecutor()
    // 创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)

 

newCachedThreadPool()	-缓存型池子，先查看池中有没有以前建立的线程，如果有，就 reuse.如果没有，就建一个新的线程加入池中 -缓存型池子通常用于执行一些生存期很短的异步型任务  因此在一些面向连接的daemon型SERVER中用得不多。但对于生存期短的异步任务，它是Executor的首选。 -能reuse的线程，必须是timeout IDLE内的池中线程，缺省     timeout是60s,超过这个IDLE时长，线程实例将被终止及移出池。   注意，放入CachedThreadPool的线程不必担心其结束，超过TIMEOUT不活动，其会自动被终止。
newFixedThreadPool(int)	-newFixedThreadPool与cacheThreadPool差不多，也是能reuse就用，但不能随时建新的线程 -其独特之处:任意时间点，最多只能有固定数目的活动线程存在，此时如果有新的线程要建立，只能放在另外的队列中等待，直到当前的线程中某个线程终止直接被移出池子 -和cacheThreadPool不同，FixedThreadPool没有IDLE机制（可能也有，但既然文档没提，肯定非常长，类似依赖上层的TCP或UDP IDLE机制之类的），所以FixedThreadPool多数针对一些很稳定很固定的正规并发线程，多用于服务器 -从方法的源代码看，cache池和fixed 池调用的是同一个底层 池，只不过参数不同: fixed池线程数固定，并且是0秒IDLE（无IDLE）     cache池线程数支持0-Integer.MAX_VALUE(显然完全没考虑主机的资源承受能力），60秒IDLE
newScheduledThreadPool(int)	-调度型线程池 -这个池子里的线程可以按schedule依次delay执行，或周期执行
newSingleThreadExecutor()	-单例线程，任意时间池中只能有一个线程 -用的是和cache池和fixed池相同的底层池，但线程数目是1-1,0秒IDLE（无IDLE）

 

    一般来说，CachedTheadPool在程序执行过程中通常会创建与所需数量相同的线程，然后在它回收旧线程时停止创建新线程，因此它是合理的Executor的首选，只有当这种方式会引发问题时（比如需要大量长时间面向连接的线程时），才需要考虑用FixedThreadPool

CompletionService & ExecutorCompletionService

public interface CompletionService<V> {
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    Future<V> take() throws InterruptedException;
    Future<V> poll();
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
<span style="font-family:Microsoft YaHei;">}</span>

    这个接口是为了方便多个任务执行时，可以方便得获取到执行任务的Future结果。同样，也是五个方法，分为两大方面。一个是对Callable和Runnable类型参数的任务提交，另一方面则是尝试对结果以不同的方式进行获取，take()方法一般是阻塞式的获取，后两者则更灵活。
    ExecutorCompletionService这个实现类主要做的就是将执行完成的任务结果放到阻塞队列中，这样等待结果的线程，如执行take()方法会得到结果并恢复执行。
    ExecutorCompletionService有3个属性：
    AbstractExecutorService类的对象aes
    Executor类的对象executor
    BlockingQueue<Future<V>>的completionQueue
    通常，如果executor是AbstractExecutorService的一个实现，则将其赋值给aes属性，否则赋值为null。
    在这个类中，executor负责执行任务，而aes则负责做适配处理，返回包装好任务的FutureTask对象。
    这里面有一个对于实现功能很重要的内部类QueueingFuture，实现如下：

private class QueueingFuture extends FutureTask {
    QueueingFuture(RunnableFuture task) {
        super(task, null);
        this.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future task;
}

主要是扩展了FutureTask的done方法，将执行结果放入BlockingQueue中

Callable

    很多场景一个任务完成执行，我们是需要知道它的结果的。为了弥补这个不足， 从JDK1.5开始，在java.util.concurrent包中就有了Callable这个接口。

 

public interface Callable<V> {
    V call() throws Exception;
}

 

    注意到，其中call()方法除了有返回结果以外，比起run()还有异常抛出，这个使用时是要注意的。

    在JavaSE5之后，执行器Executor是JDK提供给我们的良好工具，在ExecutorService中也有了支持Callable的submit()方法，那么对于其call()的执行结果我们如何来取呢，这就要提到另一个类——java.util.concurrent.Future。

Future

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask

    FutureTask是Future具体的实现类

RunnableFuture & RunnableScheduledFuture

public interface RunnableFuture extends Runnable, Future {
    void run();
}

    RunnableFuture把Runnable和Future两个接口捏到一起了。实际上，这个接口用意应该是这样的，将需要run()的任务和结果结合在一起，执行了run()能够保证结果被设置从而可以获取到。
    在ExecutorService中，submit()方法会有很多重载实现，有的用Runnable参数，有的用Callable参数。而对于submit()方法本身的实际操作，就是：执行任务和返回Future对象。
    在实现中，AbstractExecutorService的submit()方法无论传入的是Callable还是Runnable，都会调用newTaskFor()将其转变为一个RunnableFuture对象，这个对象既会被用来调用Executor的execute()方法，也会作为Future结果返回。
    JDK1.6之后，FutureTask也就成为了RunnableFuture的一个实现，当然也还是Future的实现类。我们再来简单看下它的实现。
    前面文章提到过AbstractQueuedSynchronizer类(AQS)的应用，实际上FutureTask中也有一个名为Sync而且继承于AQS的内部类。在FutureTask的实现中，每个任务都有执行状态，其巧妙地运用了AQS提供的state属性和protected方法，保证了对Future结果获取线程的阻塞和唤醒。

Atomic类

    Atomic类是通过硬件支持的CAS操作实现的，其基本思想是如果这个地址上的值和期望的值相等，则给其赋予新值，否则不做任何事儿，但是要返回原值是多少。在java 7中，共有以下几种Atomic类：
    标量类：AtomicBoolean，AtomicInteger，AtomicLong，AtomicReference
    数组类：AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray
    更新器类：AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater
    复合变量类：AtomicMarkableReference，AtomicStampedReference

    当线程数较多，竞争较为激烈的情况下，CAS效率也会受到影响。因此，在java 8中引入了LongAdder, DoubleAdder等类，他们的思路是通过分段，不同线程在不同段上做累加，最后进行汇总。

并发容器

ConcurrentMap

用分离锁实现多个线程间的更深层次的共享访问。
用 HashEntery 对象的不变性来降低执行读操作的线程在遍历链表期间对加锁的需求。
通过对同一个Volatile 变量的写/读访问，协调不同线程间读/写操作的内存可见性。

ConcurrentNavigableMap & ConcurrentSkipListMap

ConcurrentSkipListMap不同于TreeMap，前者使用SkipList（跳表）实现排序，而后者使用红黑树。相比红黑树，跳表的原理比较容易理解，简单点说就是在有序的链表上使用多级索引来定位元素。

 

ConcurrentSkiplistMap VS ConcurrentHashMap

在4线程1.6万数据的条件下，ConcurrentHashMap 存取速度是ConcurrentSkipListMap 的4倍左右。但ConcurrentSkipListMap有几个ConcurrentHashMap 不能比拟的优点：
1、ConcurrentSkipListMap 的key是有序的。
2、ConcurrentSkipListMap 支持更高的并发。ConcurrentSkipListMap 的存取时间是log(N)，和线程数几乎无关。也就是说在数据量一定的情况下，并发的线程越多，ConcurrentSkipListMap越能体现出他的优势。 
如果需要排序map，在非多线程的情况下，应当尽量使用TreeMap。此外对于并发性相对较低的并行程序可以使用Collections.synchronizedSortedMap将TreeMap进行包装，也可以提供较好的效率。对于高并发程序，应当使用ConcurrentSkipListMap，能够提供更高的并发度。

ConcurrentSkipListSet

CopyOnWriteArrayList & CopyOnWriteArraySet

队列

阻塞队列 BlockingQueue

ArrayBlockingQueue
基于数组实现，一个Lock控制互斥访问，两个condition
DelayQueue
延时队列
LinkedBlockingQueue
基于链表实现，两个锁实现
PriorityBlockingQueue
优先级队列
SynchronousQueue
同步队列
BlockingDeque & LinkedBlockingDeque

非阻塞队列 ConcurrentLinkedQueue

Lock接口

Lock & ReentrantLock

Java提供了另一种同步代码块的机制，它是基于Lock接口和它的实现类（例如ReentrantLock）来实现的，这种机制更加强大和灵活，对比Synchronized方法或者Synchronized代码块主要的优点表现在：
1）Lock接口允许更加复杂的结构，synchronized关键字必须要用结构化的方法来获取或者释放同步代码块；
2）Lock接口提供了一些额外的功能。例如tryLock()方法。
3）当只有一个写和多个读的线程时，Lock接口允许读写操作的分离
4）Lock接口的性能更高

ReadWriteLock & ReentrantReadWriteLock

    ReadWriteLock是所有Lock接口中最重要的接口之一，ReentrentReadWriteLock是它的唯一实现类。
该类有两个锁，一个是读操作另一个是写操作。它能够同时包括多个读操作，但是只能有一个写操作。当某个线程执行写操作时，其他任何线程都不能执行读操作。
Condition

    Condition 将 Object 监视器方法（wait、notify 和 notifyAll）分解成截然不同的对象，以便通过将这些对象与任意 Lock 实现组合使用，为每个对象提供多个等待 set（wait-set）。其中，Lock 替代了 synchronized 方法和语句的使用，Condition 替代了 Object 监视器方法的使用。

    Condition 实例实质上被绑定到一个锁上。要为特定 Lock 实例获得 Condition 实例，请使用其 newCondition() 方法。等待用await()，释放用signal().

Synchronized与Lock区别

synchronized是托管给JVM执行的，而lock是java写的控制锁的代码。
Lock提供了公平锁；
Lock提供了Condition；
Lock提供了取消锁等待；

synchronized和lock性能区别

    在Java1.5中，synchronize是性能低效的。因为这是一个重量级操作，需要调用操作接口，导致有可能加锁消耗的系统时间比加锁以外的操作还多。相比之下使用Java提供的Lock对象，性能更高一些。但是到了Java1.6，发生了变化。synchronize在语义上很清晰，可以进行很多优化，有适应自旋，锁消除，锁粗化，轻量级锁，偏向锁等等。导致在Java1.6上synchronize的性能并不比Lock差。官方也表示，他们也更支持synchronize，在未来的版本中还有优化余地。
说到这里，还是想提一下这2中机制的具体区别。据我所知，synchronized原始采用的是CPU悲观锁机制，即线程获得的是独占锁。独占锁意味着其他线程只能依靠阻塞来等待线程释放锁。而在CPU转换线程阻塞时会引起线程上下文切换，当有很多线程竞争锁的时候，会引起CPU频繁的上下文切换导致效率很低。
而Lock用的是乐观锁方式。所谓乐观锁就是，每次不加锁而是假设没有冲突而去完成某项操作，如果因为冲突失败就重试，直到成功为止。乐观锁实现的机制就是CAS操作（Compare and Swap）。我们可以进一步研究ReentrantLock的源代码，会发现其中比较重要的获得锁的一个方法是compareAndSetState。这里其实就是调用的CPU提供的特殊指令。
    现代的CPU提供了指令，可以自动更新共享数据，而且能够检测到其他线程的干扰，而 compareAndSet() 就用这些代替了锁定。这个算法称作非阻塞算法，意思是一个线程的失败或者挂起不应该影响其他线程的失败或挂起的算法。

同步器

信号量Semaphore

Semaphore控制同时访问资源的线程个数，如三台打印机，则只允许三个线程从job队列中取出job并执行

闭锁CountDownLatch

例如，会议室人到齐后开始会议。Conference持有一个CountDownLatch对象，设置了初始值，每个Participant持有Conference对象，当运行Participant的run方法时，调用Conference的arrive方法，从而将Conference的CountDownLatch执行countDown操作。在Conference的run方法中，执行CountDownLatch的await方法。

关卡CyclicBarrier

表示请大家等待，等所有集合都准备好了，那么就开始运行，这个过程可以循环。比如：公司部门的周末准备一起出去游玩，先等到所有的人到达汽车才开始启动车辆到目的地去，到后自由玩，然后到1点在一起吃饭。
将CyclicBarrier对象传给所有的Mapper，等所有的MapperTask执行完，将执行结果放入结果集后，调用barrier.await()方法。这时，线程会等待所有的MapperTask执行完。之后会自动调用传入CyclicBarrier的一个Runnable对象，并执行其run方法。

交换器Exchanger

用于两个线程在运行时数据交换

计时器Timer

    主要用于执行定时任务。Timer就是一个线程，使用schedule方法完成对TimerTask的调度，多个TimerTask可以共用一个Timer，也就是说Timer对象调用一次schedule方法就是创建了一个线程，并且调用一次schedule后TimerTask是无限制的循环下去的，使用Timer的cancel()停止操作。当然同一个Timer执行一次cancel()方法后，所有Timer线程都被终止。

    //以下是几种调度task的方法： 
    timer.schedule(task, time); 
    // time为Date类型：在指定时间执行一次。 
    timer.schedule(task, firstTime, period); 
    // firstTime为Date类型,period为long 
    // 从firstTime时刻开始，每隔period毫秒执行一次。 
    timer.schedule(task, delay) 
    // delay 为long类型：从现在起过delay毫秒执行一次 
    timer.schedule(task, delay, period) 
    // delay为long,period为long：从现在起过delay毫秒以后，每隔period 
    // 毫秒执行一次。

    在java正则表达式中，当遇到某些特殊的正则表达式和需要匹配的字符串时，会发生regex hang issue，例如如下程序将会导致CPU运行时间较长，如果content中的A的个数更多时，运行时间是指数级增长的。

Pattern pattern = Pattern.compile("(A*)*A");
String content = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAB";
long current = System.currentTimeMillis();
Matcher matcher = pattern.matcher(content);
if(matcher.matches()) {
    System.out.println("Matched");
} else {
    System.out.println("Not Matched");
}
System.out.println(System.currentTimeMillis()-current);

    对于上述问题，我们可以将content转换成一个字符序列CharSequence，在正则表达式匹配的时候，会隐式调用charAt()方法。因此，可以重写/覆盖charAt()方法，即每次调用的时候，加上一个标识判断，如果标识为false，则退出。另一方面，可以定义一个Timer，当达到了指定delay时间时，改变标识为false，就可以顺利退出match。具体实现如下：

public class MySequence implements CharSequence {
    private final CharSequence content;
    private final AtomicBoolean signal;
    public MySequence(CharSequence content, AtomicBoolean signal) {
        this.content = content;
        this.signal = signal;
    }
    @Override
    public int length() {
        return content.length();
    }
    @Override
    public char charAt(int index) {
        if(signal.get())
            throw new RuntimeException(new MyException("regex hang"));
        return content.charAt(index);
    }
    @Override
    public CharSequence subSequence(int start, int end) {
        return new TimeoutRegexCharSequence(content.subSequence(start, end), signal);
    }
    @Override
    public String toString() {
        return content.toString();
    }
}

public class MyTimerTask extends TimerTask {
    private volatile AtomicBoolean signal;
    public MyTimerTask(AtomicBoolean signal) {
        this.signal = signal;
    }
    @Override
    public void run() {
        signal.set(true);
    }
}

Pattern pattern = Pattern.compile("(A*)*A");
String content = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAB";
Timer timer = new Timer("name", true);
final AtomicBoolean signal = new AtomicBoolean(false);
MyTimerTask timerTask = new MyTimerTask(signal);
timer.schedule(timerTask, 10*1000);
MySequence timeoutInput = new MySequence(content, signal);
Matcher matcher;
try {
    matcher = pattern.matcher(timeoutInput);
    boolean isMatched = matcher.matches();
} catch(RuntimeException ex) {
    Throwable th = ex.getCause();
    if(th instanceof MyException) {
        System.out.println("regex hang");
    }
        throw ex;
} finally {
    timerTask.cancel();
    signal.set(false);
}