Java并发编程

来自Java并发编程的艺术
个人博客： http://blog.csdn.net/qq_22329521/article/details/52576788

并发一定比串行快么？

这个问题肯定是错的，并发比串行慢的原因在于：线程有创建和上下文切换的开销

上下文切换

即使是单核处理器也支持多线程执行代码，CPU通过给每个线程分配CPU时间片来实现这个机制。CPU通过时间片分配的算法来循环执行任务，当前任务执行一个时间片后会切换到下一个任务。但是，在切换前会保持上一个任务的状态，以便下次切换回这个任务时，可以再加之这个任务的状态。所以任务从保存到再加载的过程就是一次上下文切换。
如何减少上下文切换
--

• 无锁并发编程：多线程竞争锁，会引起上下文切换，所以多线程处理数据时，可以用一些办法避免使用锁，如将数据的ID按照Hash算法取模分段，不同的线程处理不同段的数据
• CAS算法：Java的Atomic包使用CAS算法来更新数据，而不需要加锁。
• 使用最少线程：避免创建不需要的线程，比如任务很少，但是创建了很多线程来处理，这样会造成大量线程处于等待
• 协程：在单行线程中实现多任务调度，并在单线程中维持多个任务的切换

避免死锁的几种方式

• 避免一个线程同时获取多个锁
• 避免一个线程在锁内同时占用多个资源，尽量保证每个锁只占用一个资源
• 尝试使用定时锁，使用lock.tryLock(timeout)来替代使用内部锁机制。
• 对于数据库锁，加锁和解锁必须在一个数据库连接中里，否则会出现解锁失败的情况。

资源限制

资源限制指的是程序的执行速度受限于计算机硬件资源或软件资源，如服务器的带宽只有2Mb/s,某个资源的下载速度为1Mb/s,系统启动10个线程去下载资源，下载速度不会变成10Mb/s，所以在进行并发的时候回考虑资源的限制。硬件资源限制有带宽的上传/下载速度、硬盘的读写速度和CPU的处理速度。软件资源限制有数据库的连接数和socket连接数等。

资源限制引来的问题：为了将代码执行速度加快将代码中串行执行的部分变成并发执行，因为资源受限，仍然在串行执行，这时候程序不仅不会加快，反而会变慢，因为增加了上下文切换和资源调度的时间。

如何解决资源限制问题：可以使用集群并行执行程序，既然单机的资源有限，那么可以让程序在多机上运行，比如使用ODPS、Hadoop或者自己搭个服务器集群，不同的机器处理不同的数据，可以通过“数据ID%机器数”，计算得到一个机器编号，然后由对应编号的机器处理这个数据，对于软件资源受限，可以使用资源池来复用如使用连接池将数据库和Socket连接复用，或者在调用对方webservice接口获取数据只建立一个连接。

Java并发机制的底层实现原理

Java代码在编译后会变成Java字节码，字节码被类加载器加载到JVM里，JVM执行字节码，最终需要转化为汇编指令在CPU上执行，Java所使用的并发机制依赖于JVM的实现和CPU的指令

volatile的应用

volatile是轻量级的synchronized，在多处理器并发中保证了共享变量的可见性，可见性是指当一个线程修改了一个共享变量，另一个线程能读到修改的值，它不会引起线程上下文切换和调度

volatile在java代码转换为汇编代码 会多了一个Lock前缀的指令，在多核处理器下发生两件事情

• 将当前处理器缓存行的数据写回到系统内存
• 将这个协会内存的操作会使得其他CPU力缓存了该内存的地址的数据无效

为了提高处理速度，处理器不直接和内存通信，而是将系统内存的数据读到内部缓存（L1，L2或其他）后再进行操作，但操作完不知道何时回写到内存，如果声明了volatile的变量进行写操作，JVM就会向处理器发送一条Lock前缀的指令，将这个变量所在缓存行的数据写回到系统内存，但是就是写会内存，如果其他处理器缓存的值还是旧的，再执行计算操作就会有问题，所以在多处理器下为了保证各个处理器的缓存是一致，就会执行缓存一致性协议，每个处理器通过嗅探在总线上传播的数据来检查自己的缓存值是否过期，当处理器发现自己缓存行所对应的内存地址被修改，就会将当前处理器缓存行设置为无效，当处理器对数据进行修改操作，会重新从系统内存读到处理器缓存中

synchronized的和应用

javase1.6 对synchronized进行各种优化，过去被人称为重量级锁。
java每个对象都是锁

• 对于普通同步方法，锁就是实例对象
• 对于静态同步方法，锁就是当前类的Class对象
• 对于同步代码块，锁就是Synchonized括号里配置的对象

JVM基于进入和退出Monitor对象来实现方法同步和代码块同步
synchronized用的锁是存在Java对象头里的。Java对象头里的Mark Word力默认储存对象的HashCode，分代年龄和锁标记位。在运行期间，MarkWord储存的数据会随着锁标记位的变化而变化

Javase1.6中 锁一共有4种状态，级别从低到高为：无锁状态、偏向锁状态、轻量级锁状态和重量级锁状态，这几个状态会随着竞争情况逐渐升级。锁可以升级但不能降级，目的是为了提高获得锁和释放锁的效率

1. 偏向锁：大多数情况下，锁不仅不存在多线程竞争，而且总是同一个线程多次获取，为了让线程获得锁的代价更低引入偏向锁。当某一线程访问同步块时，会在对象头和栈帧中的琐记录里存储锁偏向的线程ID，以后该线程在进入该同步块的时候，不需要再次使用CAS原子操作进行加锁和解锁，只需要简单的测试一下对象头中的Mark Word是否存在指向当前线程的偏向锁。如果测试成功，则表示获得锁，否则检测是否设置有偏向锁，如果没有，则使用CAS竞争锁，否则偏向锁指向该线程。
2. 轻量级锁：线程执行同步块之前，会在线程私有的栈帧中开辟用于存储锁记录的空间，称为Displaced Mark Word。然后线程尝试将对象Mark Word的替换为指向Displaced Mark Word记录的指针，如果成功，那么当前线程获得锁，如果失败，那么使用自旋获得锁。

原子操作的实现原理

1. 处理器实现原子操作：使用总线锁保证原子性，使用缓存锁保证原子性（修改内存地址，缓存一致性机制：阻止同时修改由2个以上的处理器缓存的内存区域数据）
2. JAVA实现原子操作：循环使用CAS实现原子操作

public class Counter {
    private AtomicInteger atomicI = new AtomicInteger(0);
    private int i = 0;
    public static void main(String[] args) {
        final Counter cas = new Counter();
        List<Thread> ts = new ArrayList<Thread>(600);
        long start = System.currentTimeMillis();
        for (int j = 0; j < 100; j++) {
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 10000; i++) {
                        cas.count();
                        cas.safeCount();
                    }
                }
            });
            ts.add(t);
        }

        for (Thread t : ts) {
            t.start();
        }
        // 等待所有线程执行完成
        for (Thread t : ts) {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        System.out.println(cas.i);
        System.out.println(cas.atomicI.get());
        System.out.println(System.currentTimeMillis() - start);

    }

    /**
     * 使用CAS实现线程安全计数器
     */
    private void safeCount() {
        for (;;) {
            int i = atomicI.get();
            boolean suc = atomicI.compareAndSet(i, ++i);
            if (suc) {
                break;
            }
        }
    }
    /**
     * 非线程安全计数器
     */
    private void count() {
        i++;
    }
}

CAS实现原子操作的三大问题

1. ABA问题：因为CAS需要在操作值的时候检查下值有没有发生变化，如果没有发生变化则更新，但是如果一个值原来是A，变成了B，又变成了A，那么使用CAS进行检查时会发现它的值没有发生变化，但是实际上却变化了。ABA问题的解决思路就是使用版本号。在变量前面追加上版本号，每次变量更新的时候把版本号加一，那么A－B－A 就会变成1A-2B－3A。
2. 循环时间长开销大。自旋CAS如果长时间不成功，会给CPU带来非常大的执行开销。如果JVM能支持处理器提供的pause指令那么效率会有一定的提升，pause指令有两个作用，第一它可以延迟流水线执行指令（de-pipeline）,使CPU不会消耗过多的执行资源，延迟的时间取决于具体实现的版本，在一些处理器上延迟时间是零。第二它可以避免在退出循环的时候因内存顺序冲突（memory order violation）而引起CPU流水线被清空（CPU pipeline flush），从而提高CPU的执行效率。
3. 只能保证一个共享变量的原子操作。当对一个共享变量执行操作时，我们可以使用循环CAS的方式来保证原子操作，但是对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁，或者有一个取巧的办法，就是把多个共享变量合并成一个共享变量来操作。比如有两个共享变量i＝2,j=a，合并一下ij=2a，然后用CAS来操作ij。从Java1.5开始JDK提供了AtomicReference类来保证引用对象之间的原子性，你可以把多个变量放在一个对象里来进行CAS操作。

Java内存模型

内容较多涉及到内存模型、重排序
http://blog.csdn.net/ccit0519/article/details/11241403（内容较多也比较详细介绍）
double_check的问题

public class DoubleCheckedlocking{
  private static Instance instance;
  public static Instance getInstance(){
     if(instance==null){
      synchronized(DoubleCheckedlocking.class){
       if(instance==null)
         instance=new Instance();
       }
     }
  }
}
//根据重排序可能会出现的问题
instance=new Instance()常见一个对象可以分成三步
memory=allocate(),//1.分配对象的内存空间
ctorInstance（memory)//2.初始化对象
instance=memory //3.设置Instance指向刚分配的内存地址
如果2,3 重排序颠倒后  if语句就可以是引用是上存在但是对象还未被初始化，所以 可以给Instance加上一个volatile因为内存屏障的缘故

Java中的锁

锁是用来控制多个线程访问共享资源的方式，一般来说，一个锁能够防止多个线程同时访问共享资源，但是有些锁可以运行多个线程并发的访问共享资源，比如读写锁。Lock接口和synchronized可以通过获取锁和释放锁，但是前者比后者更具扩展性。
Lock是一个接口，定义了锁获取和释放的基本操作
Lock和synchronized区别

• Lock接口可以尝试非阻塞地获取锁 当前线程尝试获取锁。如果这一时刻锁没有被其他线程获取到，则成功获取并持有锁。
• Lock接口能被中断地获取锁 与synchronized不同，获取到锁的线程能够响应中断，当获取到的锁的线程被中断时，中断异常将会被抛出，同时锁会被释放。
• Lock接口在指定的截止时间之前获取锁，如果截止时间到了依旧无法获取锁，则返回。

Lock接口的APi

• void lock() 获取锁,调用该方法当前线程将会获取锁，当锁获取后，该方法将返回。
• void lockInterruptibly() throws InterruptedException 可中断获取锁，与lock()方法不同之处在于该方法会响应中断，即在锁的获取过程中可以中断当前线程
• boolean tryLock() 尝试非阻塞的获取锁，调用该方法立即返回，true表示获取到锁
• boolean tryLock(long time,TimeUnit unit) throws InterruptedException 超时获取锁，以下情况会返回：时间内获取到了锁，时间内被中断，时间到了没有获取到锁。
• void unlock() 释放锁

队列同步器（AQS）

队列同步器AbstractQueuedSynchronizer（以下简称同步器），是用来构建锁或者其他同步组件的基础框架，它使用了一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。同步器的主要使用方式是继承，子类通过继承同步器并实现它的抽象方法来管理同步状态，在抽象方法的实现过程中免不了要对同步状态进行更改，这时就需要使用同步器提供的3个方法（getState()、setState(int newState)和compareAndSetState(int expect,int update)）来进行操作，因为它们能够保证状态的改变是安全的。同步器既可以支持独占式地获取同步状态，也可以支持共享式地获取同步状态，这样就可以方便实现不同类型的同步组件（ReentrantLock、ReentrantReadWriteLock和CountDownLatch等）。同步器是实现锁（也可以是任意同步组件）的关键，在锁的实现中聚合同步器，利用同步器实现锁的语义。可以这样理解二者之间的关系：锁是面向使用者的，它定义了使用者与锁交互的接口（比如可以允许两个线程并行访问），隐藏了实现细节；同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。锁和同步器很好地隔离了使用者和实现者所需关注的领域

enter image description here

enter image description here

public class Mutex implements Lock {
    private static class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) throw new IllegalMonitorStateException();
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    private final Sync sync = new Sync();

    @Override
    public void lock() {
        sync.acquire(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(time));
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }

}

以上是独占式锁是一个自定义的同步组件，在同一个时刻指运行一个线程占有锁，用户在使用Mutex并不会直接和内部同步器打交道，而是调用Mutex提供的方法，在Mutex的实现中，获取锁Lock方法。
同步队列
--
同步器依赖内部的同步队列（一个FIFO双向队列）来完成同步状态的管理。同步队列中的节点（Node）用来保存"获取同步状态失败的线程"引用、等待状态以及前驱和后继节点。

enter image description here

enter image description here

当前线程获取同步状态失败时，同步器会将当前线程、等待状态等信息构造成为一个节点（Node）并将其加入同步队列，同时会”“阻塞”当前线程。当一个线程成功地获取了同步状态（或者锁），其他线程将无法获取到同步状态，转而被构造成为节点并加入到同步队列中，而这个加入队列的过程必须要保证线程安全。同步器提供了一个基于CAS的设置尾节点的方法：compareAndSetTail(Nodeexpect,Nodeupdate)，它需要传递当前线程“认为”的尾节点和当前节点，只有设置成功后，当前节点才正式与之前的尾节点建立关联。

独占式同步状态获取与释放

主要逻辑：首先调用自定义同步器实现的tryAcquire(int arg)方法，该方法保证线程安全的获取同步状态，如果同步状态获取失败，则构造同步节点（独占式Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWaiter(Node node)方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态

enter image description here

//将节点加入到同步队列的尾部
private Node addWaiter(Node mode) {
      Node node = new Node(Thread.currentThread(), mode);//“生成节点”
      // Try the fast path of enq; backup to full enq on failure
      //快速尝试在尾部添加
      Node pred = tail;
      if (pred != null) {
          node.prev = pred;//先将当前节点node的前驱指向当前tail
          if (compareAndSetTail(pred, node)) {//CAS尝试将tail设置为node
              //如果CAS尝试成功，就说明"设置当前节点node的前驱"与"CAS设置tail"之间没有别的线程设置tail成功
              //只需要将"之前的tail"的后继节点指向node即可
              pred.next = node;
              return node;
          }
      }
      enq(node);//否则，通过死循环来保证节点的正确添加
      return node;
  }
private Node enq(final Node node) {
    for (;;) {//通过死循环来保证节点的正确添加
        Node t = tail;
        if (t == null) { // Must initialize 同步队列为空的情况
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {//直到CAS成功为止
                t.next = node;
                return t;//结束循环
            }
        }
    }
}

上述代码通过使用compareAndSetTail（Node expect,Node update)来确保节点能够被线程安全添加，如果使用普通的LinkedList来维护节点之间的关系，那么当一个线程获取到同步状态，而其他多个线程由于调用tryAcquire（int arg）方法获取同步状态失败而并发被添加到LinkedList，LinkedList将难以保证Node的正确添加
在enq(final Node node)方法中，同步器通过“死循环”来保证节点的正确添加，在“死循环”中只有通过CAS将节点设置成为尾节点之后，当前线程才能从该方法返回，否则，当前线程不断地尝试设置。可以看出，enq(final Node node)方法将并发添加节点的请求通过CAS变得“串行化”了。
节点自旋
--
节点进入同步队列之后，就进入了一个自旋的过程，每个节点（或者说是线程）都在自省地观察，当条件满足，获取到了同步状态，就可以从这个自旋过程中退出，否则依旧留在这个自旋过程中。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {//无限循环
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {//前驱节点是首节点且获取到了同步状态
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;//从自旋中退出
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//判断获取同步状态失败后是否需要阻塞
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

共享式同步状态获取与释放

共享式获取与独占式获取最主要的区别在于同一时刻能否有多个线程同时获取到同步状态。
以文件的读写为例，如果一个程序在对文件进行读操作，那么这一时刻对于该文件的写操作均被阻塞，而读操作能够同时进行。写操作要求对资源的独占式访问，而读操作可以是共享式访问

enter image description here

调用同步器的acquireShared(int arg)方法可以共享式地获取同步状态。

  public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
      private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

在acquireShared(int arg)方法中，同步器调用tryAcquireShared(int arg)方法尝试获取同步状态，tryAcquireShared(int arg)方法返回值为int类型，当返回值大于等于0时，表示能够获取到同步状态。因此，在共享式获取的自旋过程中，成功获取到同步状态并退出自旋的条件就是tryAcquireShared(int arg)方法返回值大于等于0。
在doAcquireShared(int arg)方法的自旋过程中，如果当前节点的前驱为头节点时，尝试获取同步状态，如果返回值大于等于0，表示该次获取同步状态成功并从自旋过程中退出。
重入锁
--
它表示该锁能够支持一个线程对资源的重复加锁。除此之外，该锁的还支持获取锁时的公平和非公平性选择
之前的例子，当一个线程调用Mutex的lock()方法获取锁之后，如果再次调用lock()方法，则该线程将会被自己所阻塞，原因是Mutex在实现tryAcquire(int acquires)方法时没有考虑占有锁的线程再次获取锁的场景，而在调用tryAcquire(int acquires)方法时返回了false，导致该线程被阻塞。简单地说，Mutex是一个不支持重进入的锁。而synchronized关键字隐式的支持重进入，比如一个synchronized修饰的递归方法，在方法执行时，执行线程在获取了锁之后仍能连续多次地获得该锁，而不像Mutex由于获取了锁，而在下一次获取锁时出现阻塞自己的情况。
ReentrantLock虽然没能像synchronized关键字一样支持隐式的重进入，但是在调用lock()方法时，已经获取到锁的线程，能够再次调用lock()方法获取锁而不被阻塞。
公平锁与非公平锁的比较
--
公平性锁每次都是从同步队列中的第一个节点获取到锁，而非公平性锁出现了一个线程连续获取锁的情况。
非公平性锁可能使线程“饥饿”，当一个线程请求锁时，只要获取了同步状态即成功获取锁。在这个前提下，刚释放锁的线程再次获取同步状态的几率会非常大，使得其他线程只能在同步队列中等待。
为什么它又被设定成默认的实现呢？非公平性锁模式下线程上下文切换的次数少，因此其性能开销更小。公平性锁保证了锁的获取按照FIFO原则，而代价是进行大量的线程切换。非公平性锁虽然可能造成线程“饥饿”，但极少的线程切换，保证了其更大的吞吐量。

读写锁

读写锁在同一时刻可以允许多个读线程访问，但是在写线程访问时，所有的读线程和其他写线程均被阻塞。
读写锁维护了一对锁，一个读锁和一个写锁，通过分离读锁和写锁，使得并发性相比一般的排他锁有了很大提升。除了保证写操作对读操作的可见性以及并发性的提升之外，读写锁能够简化读写交互场景的编程方式。在读多于写的情况下，读写锁能够提供比排它锁更好的并发性和吞吐量。Java并发包提供读写锁的实现是ReentrantReadWriteLock。

ConcurrentHashMap的实现原理与使用

ConcurrentHashMap是线程安全且高效的HashMap。
为什么使用ConcurrentHashMap的原因

1. HashMap线程不安全，在多线程下使用HashMap进行put操作会引起死循环，导致CPU利用率接近100%，原因在于多线程会导致HashMap的Entry链表形成环形数据结构，一旦形成环形数据结构，Entry的next节点永远不为空，就回产生死循环获取Entry
2. HashTable效率低
3. ConcurrentHashMap的锁分段技术可以有效提升并发访问率，原因在于HashTable在竞争中都是竞争同一把锁，但是ConcurrentHashMap将数据分成一段一段地储存，然后给每一段数据配一把锁，当一个线程占用锁访问其中一段数据时候，其他段的数据也被其他线程访问
   具体的实现及原理http://www.ibm.com/developerworks/cn/java/java-lo-concurrenthashmap/

Fork/Join框架

Fork/Join框架是Java7提供了的一个用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

enter image description here