Java多线程整理（li）

时间 2019-11-06

标签 java 多线程整理栏目 Java 繁體版

原文原文链接

目录：html

1.volatile变量java

2.Java并发编程学习算法

3.CountDownLatch用法编程

4.CyclicBarrier使用api

5.BlockingQueue使用数组

6.任务执行器Executor
7.CompletionService使用
8.ConcurrentHashMap使用
9.Lock使用缓存

1、 volatile变量安全

　　1.volatile原理：volatile的原理其实是告诉处理器，不要把变量缓存在寄存器或者相对于其余处理器不可见的地方，而是把变量放在主存，每次读写操做都在主存上进行操做。另外，被申明为volatile的变量也不会与其它内存中的变量进行重排序。
服务器

　　2.volatile同步：volatile是同步的一个子集，只保证了变量的可见性，可是不具有原子特性。这就是说线程可以自动发现 volatile 变量的最新值。相对于同步而言，volatile的优点：a.简易性，能够像使用其余变量同样使用volatile变量；b.volatile变量不会形成线程阻塞；c.若是读操做远远大于写操做，volatile 变量还能够提供优于锁的性能优点。网络

　　3.正确使用volatile条件：对变量的写操做不依赖于当前值；该变量没有包含在具备其余变量的不变式中；

/*
 * 对于第一条原则:对变量的写操做不依赖于当前值;
 * 虽然i++只有一条语句,实际上这条语句是分三步执行的,读入i,i加1，写入i;
 * 若在第三步执行过程前，其余线程对i进行了改动，此时的结果将是错的。所以即便使用了volatile进行控制，并不能保证这个操做是线程安全的。
*/

private volatile int i=1;
... 
i++;

/*
 * 这类问题的解决方案有两种: 
 * 1.一种是采用synchronized进行同步控制，这显然违背了volatile的初衷
 * 2.一种是采用CPU原语进行控制。在jdk1.5以后，java.util.concurrent.atomic包下的不少类就是采用这种方式进行控制，这样能够在保持性能的状况下，保证数据的线程安全。
 */

2、Java并发编程学习

　　在java 并发编程实践中对线程安全的定义以下：当多个线程访问一个类时，若是不用考虑这些线程在运行时环境下的调度和交替运行，而且不须要额外的同步及在调用方代码没必要作其余的协调，这个类的行为仍然是正确的，那么这个类就是线程安全的。彻底由线程安全的类构成的程序不必定就是线程安全的。

/*
 * 以下代码所示: 
 * 虽然Vector是一个线程安全的类，可是对于由Vector线程安全的方法组成上面的逻辑，显然不是线程安全的。 
 * 固然，由线程不安全的类构成的程序，也不必定不是线程安全的。
 */

Vector v;
 
if(!v.contains(o)){
v.add(o);
}

/*
 * 微妙的可见性:
 * 当变量的读入写出被不一样的线程共享时，必须使用同步。若不使用同步将有可能发生与直觉截然不同的错误。
 */
public class TestVisiable {

    static int x = 0, y = 0;
    static int b = 0, a = 0;

    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(new Runnable() {

            @Override
            public void run() {
                a = 1;
                x = b;
            }
        });
        Thread t2 = new Thread(new Runnable() {

            @Override
            public void run() {
                b = 1;
                y = a;
            }
        });
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(x + " " + y);
    }
}

/*
 * 因为没有同步，程序运行的可能结果为(1,0) (0,1),(1,1),然而，还有可能为(0,0),难以想象吧。 
 * 1.这是因为为了加快并发执行，JVM内部采用了重排序的机制致使。
 * 2.最低限的安全性在java中，java存储模型要求java变量的获取和存储都是原子操做，可是有两种类型的变量是比较特殊的,没有申明为volitale的long和double变量。 
 * 缘由在于：
 * 1.在java中long和double变量为64位，jvm容许将64位的读写操做划分为两个32位的操做。
 * 2.当多个线程同时读写非volatile的long或者double类型数据时，将有可能获得数据是一个数的高32位和另外一个数的低32位组成的数。
 * 3.这样的结果显然是彻底不对的。所以对应long或double类型的数据用于多线程共享时，必须申明加上volatile或者进行同步。
 */

3、CountDownLatch用法

　　在jdk API中以下描述：一个同步辅助类，在完成一组正在其余线程中执行的操做以前，它容许一个或多个线程一直等待。用给定的计数初始化 CountDownLatch。因为调用了 countDown() 方法，因此在当前计数到达零以前，await 方法会一直受阻塞。以后，会释放全部等待的线程，await 的全部后续调用都将当即返回。

　　CountDownLatch 是一个通用同步工具，它有不少用途。将计数 1 初始化的 CountDownLatch 用做一个简单的开/关锁存器，或入口：在经过调用 countDown() 的线程打开入口前，全部调用 await 的线程都一直在入口处等待。

/*
 * 下面给出一个详细的应用场景 百米赛跑：
 * 比赛共有10名运动员参与，10名运动在起跑线上统一等待号令员发起起跑的号令，经过终点时号令员统计该运动员的时间，当全部运动员都跑到终点时，报告每一个人的成绩。
 * 这里实际上就能够采用CountDownLatch来解决起跑线上设置一个CountDownLatch，当号令员没有发起起跑号令时，全部运动员都在起跑线上等待。
 * 在终点设置一个CountDownLatch，起跑后号令员一直等待，当全部运动员都经过终点后，它会报告成绩。 用代码实现以下：
 */

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

public class Commander {

    private final CountDownLatch     startSignal;                            // 起跑信号
    private final CountDownLatch     endSignal;                              // 终点信号
    private Long                     startTime;
    private final Map<Integer, Long> scoreMap = new HashMap<Integer, Long>();

    public void waitStart() {
        try {
            startSignal.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public synchronized void start() {
        startTime = System.currentTimeMillis();
        startSignal.countDown();
        try {
            endSignal.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void reach(int id) {
        endSignal.countDown();
        scoreMap.put(id, System.currentTimeMillis() - startTime);// 统计时间
    }

    public Commander(int num){
        startSignal = new CountDownLatch(1);
        endSignal = new CountDownLatch(num);
    }

    public static void main(String[] args) {
        int runnerNum = 10;
        Commander c = new Commander(runnerNum);
        for (int i = 0; i < runnerNum; i++) {
            new Thread(new Runner(i + 1, c)).start();
        }
        c.start(); // 发起号令
        for (Integer i : c.scoreMap.keySet()) {
            System.out.println(i + "号运动员,耗时" + c.scoreMap.get(i));
        }
    }
}

class Runner implements Runnable {

    Commander commander;
    int       id;

    public Runner(int id, Commander commander){
        this.id = id;
        this.commander = commander;

    }

    @Override
    public void run() {
        commander.waitStart();
        try {
            Thread.sleep((long) (Math.random() * 1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        commander.reach(id);
    }
}

4、CyclicBarrier使用

　　CyclicBarrier在jdk中描述：一个同步辅助类，它容许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 颇有用。由于该 barrier 在释放等待线程后能够重用，因此称它为循环的 barrier。

　　实际上CyclicBarrier相似于CountDownLatch也是个计数器。不一样的是，当线程调用await方法后，必须全部线程都到达了，才能分别进入后面的执行过程。一旦有一个线程未到达，全部线程都会等待，有点像一部电影的名字《一个都不能少》。

/*
 * 考虑一个应用场景： 
 * 老师带领学生去春游，下午回来时，须要整队，而后统一坐车回去。 
 * 在这里，老师和已经到达的学生，必须等待全部学生归队后才能回去。 用代码实现以下：
 */

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class Teacher {

    private final CyclicBarrier barrier = new CyclicBarrier(20, new Runnable() {

                                            @Override
                                            public void run() {
                                                System.out.println("全部学生已经归队");
                                            }
                                        });

    public void comeBack(int i) {
        System.out.println(i + "已经归队。");
        try {
            barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        Teacher t = new Teacher();
        for (int i = 0; i < 20; i++) {
            new Thread(new Student(i + 1, t)).start();
        }

    }
}

class Student implements Runnable {

    private final int     id;
    private final Teacher teacher;

    public Student(int id, Teacher teacher){
        this.id = id;
        this.teacher = teacher;
    }

    @Override
    public void run() {
        try {
            Thread.sleep((long) (Math.random() * 1000));
            teacher.comeBack(id);
            System.out.println(id + "上车");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

5、BlockingQueue 学习

　　从名字上看，BlockingQueue是阻塞队列的意思。这个队列主要提供下面的功能：

　　1.阻塞队列提供了可阻塞的take和put方法，另外可定时的poll和offer实际原理也是同样的。

　　2.若是BlockingQueue是空的，从BlockingQueue取东西的操做将会被阻断进入等待状态，直到BlockingQueue进了东西才会被唤醒，一样，若是BlockingQueue是满的，任何试图往里存东西的操做也会被阻断进入等待状态，直到BlockingQueue里有空间时才会被唤醒继续操做。

　　阻塞队列有通常又为两类：无限阻塞队列和有限阻塞队列。有限阻塞队列中，当队列满时，调用put方法将会阻塞；而无限阻塞队列中，put方法是不会阻塞的。很显然这个队列的一种最经常使用的场景就是：生产者-消费者模式。

/* 它有如下具体实现类：
 * ArrayBlockingQueue：采用数组做为存储队列的阻塞队列，这个队列采用FIFO的方式管理数据
 * LinkedBlockingQueue：采用链式结构做为存储队列，一样它也采用FIFO的方式管理数据
 * PriorityBlockingQueue：采用基于优先级堆的极大优先级队列做为存储队列。
 * SynchronousQueue：特殊的BlockingQueue，其中每一个 put 必须等待一个 take，反之亦然。
 * DelayQueue：这是一个无限阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部 是延迟期满后保存时间最长的 Delayed 元素。若是延迟都尚未期满，则队列没有头部。
 * 
 * 经过查看源代码能够看到这个类内部是采用PriorityQueue做为存储队列
 * DelayQueue的一些经常使用的场景
 * a) 关闭空闲链接。服务器中，有不少客户端的链接，空闲一段时间以后须要关闭之。
 * b) 缓存。缓存中的对象，超过了空闲时间，须要从缓存中移出。
 * c) 任务超时处理。在网络协议滑动窗口请求应答式交互时，处理超时未响应的请求。
 * */

6、任务执行器Executor

　　记得在大学的时候，有一次写程序时，须要建立不少的线程来处理各类socket请求，因而写了一个线程类，每出现一个socket请求，就建立一个线程。后来，老师指出，每次建立线程的开销比较大，能够将线程与具体的业务逻辑分离开来，而后用一个队列，保存必定量的线程，每次须要的时候就去取，不用的时候就还回去，这样能够循环使用，避免重复建立线程的开销，因而乎，对代码进行重构，写了大段的代码，发现之后须要用到多线程的地方均可要用它，后来在网上找资料才知道，那叫线程池。

　　jdk1.5的升级，给咱们带来一个很特殊的包java.util.concurrent,翻阅API，能够看到jdk中已经封装了线程池，简单两行即可实现。这个包中提供了Executor的接口，接口定义以下：

void execute(Runnable command);

传入一个runnable接口的实现，它便自动为咱们建立线程和执行，任务的提交者不再用为了并发执行，本身写一大段代码来建立并管理各类线程。扩展了Executor的有ExecutorService, ScheduledExecutorService,AbstractExecutorService, ScheduledThreadPoolExecutor, ThreadPoolExecutor.因为Runnable接口中只提供了一个不带返回值run方法，所以当任务须要返回值时，Executor就不能知足需求了，因而出现了ExecutorService,这个接口继承了Executor，对提交任务的接口进行了扩展，引入了Callable接口，该接口定义以下：

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

同时接口将任务执行过程进行管理，分为三个状态，提交，shutdown，terminate。在AbstractExecutorService中能够看到submit(Callable c)的实现，实际上它会先建立一个Future对象，而后再调用execute(Runnable command)方法，执行任务。能够很明显的知道Future确定是继承了Runnable接口。经过Future接口，咱们能够获取由call方法调用的返回值。Executor接口的两个具体实现是ThreadPoolExecutor和ScheduledThreadPoolExecutor，经过名字能够看出，ThreadPoolExecutor是经过采用线程池来执行每一个提交的任务，ScheduledThreadPoolExecutor继承自ThreadPoolExecutor,主要用于知足延迟后运行任务，或者按期执行任务的需求。虽然，咱们能够经过直接调用ThreadPoolExecutor和ScheduledThreadPoolExecutor的构造函数生成Executor，可是不少状况下，没有这个必要，由于jdk为咱们做了简化工做，经过Executors这个工厂类，能够只需传入一些简单的参数，即可以获得咱们须要的Executor对象。

7、CompletionService学习

　　若在采用Executor执行任务时，可用经过采用Future来获取单个任务执行的结果，在Future中提供了一个get方法，该方法在任务执行返回以前，将会阻塞。当向Executor提交批处理任务时，而且但愿在它们完成后得到结果，若是用FutureTask，你能够循环获取task，并用future.get()去获取结果，若没有完成则阻塞，这对于对任务结果须要分别对待的时候是可行的。可是若全部task产生的结果均可以被同等看待，这时候采用前面这样的方式显然是不可行了，由于若当前的task没有完成，然后面的其它task已经完成，你也得等待，这个实效性不高。显然不少时候，统一组task产生的结果都应该是没有区别的，也就是知足上述第二种状况。这个时候咋办呢？jdk为咱们提供了一个很好的接口CompletionService，这个接口的具体实现类是ExecutorCompletionService。该类中定义下面三个属性：

private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;

executor由构造函数传入，aes只是用于生成Future对象。特别要注意是completionQueue。它维护了一个保存Future对象的BlockingQueue。当这个Future对象状态是结束的状态的时候，也就是task执行完成以后，会将它加入到这个Queue中。究竟是在哪里将完成的Future加入到队列里面的呢？又是怎么知道task是何时结束的呢？在ExecutorCompletionService中定义了一个QueueingFuture类，该类的实现：

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

能够看到在done方法中，它会把当前的task加入到阻塞队列中。追踪done方法能够看到，该方法定义在FutureTask中，默认实现为空，从注释能够看出，当Future的状态转为isDone的时候，就会调用该方法。调用端在调用CompletionService的take方法时，实际上调用的是BlockingQueue的take方法，由前面的学习中，咱们知道，当队列中有内容时，该队列会当即返回队列中的对象，当队列为空时，调用线程将会阻塞。而只要有任务完成，调用线程就会跳出阻塞，得到结果。

8、ConcurrentHashMap学习

　　java中提供对HashMap是咱们使用得比较多的一个类，单该类是非线程安全的，若处于多线程环境中，则须要经过synchronized关键字进行同步（经过查看Collections.synchronizedMap方法，能够知道，实际上该方法对实现也是经过synchronized关键字进行同步控制），虽然它能保证线程安全，可是，因为须要对整个map进行加锁，这样作的并发性能每每不是很理想，尤为是map中数据量比较大的时候。jdk1.5以后，java.util.concurrent包中提供了ConcurrentHashMap，一方面，该类本身保证了线程安全性，另外一方面，该类也提供了一些复合操做的原子性接口。

　　ConcurrentHashMap与HashMap同样，一样是哈希表，可是采用不一样对锁机制--分离锁，即采用不一样的锁来同步不一样的数据块，以减小对锁对竞争。　　在ConcurrentHashMap内部采用了一个包含16个锁对象对数组，每一个锁负责同步hash Bucket的1/16，bucket中的每一个对象经过它的hashCode计算获得它所在锁对象。假设hash算法的实现可以提供合理的扩展性,而且关键字可以以统一的方式访问,这会将对于锁的请求减小到原来的1/16.这种基于分离锁设计的技术实现可以使得ConcurrentHashMap支持16个并发的请求。

9、Lock使用

　　在java5.0之前都是采用synchronized关键字进行同步控制，全部对象都自动含有单一的锁，JVM负责跟踪对象被加锁的次数。若是一个对象被解锁，其计数变为0。在任务（线程）第一次给对象加锁的时候，计数变为1。每当这个相同的任务（线程）在此对象上得到锁时，计数会递增。只有首先得到锁的任务（线程）才能继续获取该对象上的多个锁。每当任务离开一个synchronized方法，计数递减，当计数为0的时候，锁被彻底释放，此时别的任务就可使用此资源。因为这些锁是由JVM来控制，所以也叫隐式锁。

　　在jdk5.0之后，提供了显示锁，即Lock，能够说是对隐式锁的功能的扩展，主要有两个。一个是ReentrantLock,另外一个是ReentrantReadWriteLock，前者是普通重入锁，后者是可重入的读写锁。这些锁提供了两种锁竞争机制：公平竞争和非公平竞争，公平竞争的实现其实是采用一个队列保存等待的线程，当当前线程释放锁以后，取出队头的线程唤醒，使之能够获取锁。java中Lock接口的定义:

public interface Lock {
    void lock();
    void lockInterruptibly() throws InterruptedException;
    boolean tryLock();
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    void unlock();
    Condition newCondition();
}

Lock与synchronized的区别：

1.synchronized是在JVM层面上实现的，不但能够经过一些监控工具监控synchronized的锁定，并且在代码执行时出现异常，JVM会自动释放锁。而Lock的释放必须由程序本身保证，经常使用的写法是把是把释放锁的代码写到try{}finally中，在finally块中释放锁。

2.若是使用 synchronized ，若是A不释放，B将一直等下去，不能被中断，若是使用Lock，若是A不释放，可使B在等待了足够长的时间之后，中断等待，而干别的事情。

3.synchronize其实是ReentrantLock和Condition的组合的简化版

4.相对于synchonized独占锁，ReentrantReadWriteLock经过分离读锁和写锁，提供可共享的读锁提升读并发性能。