Java多线程&高并发

时间  2019-12-06
标签 java 多线程 并发 栏目 Java 繁體版
原文   原文链接

1、线程安全性

定义：当多个线程访问某个类时，无论运行时环境采用 何种调度方式，或者这些线程将如何交替执行，而且在主调代码中 不须要任何额外的同步或协同，这个类都能表现出 正确的行为，那么就称这个类是线程安全的。

1. 原子性：提供了互斥访问，同一时刻只能有一个线程来对它进行访问。html

Atomic包：编程

  1. AtomicXXX：CAS、Unsafe.compareAndSwapInt
  2. AtomicLong、LongAdder
  3. AtomicReference、AtomicReferenceFieldUpdater
  4. AtomicStampReference：CAS的ABA问题

原子性 - synchronized（同步锁）
修饰代码块：大括号括起来的代码，做用于调用的对象
修饰方法：整个方法，做用于调用的对象
修饰静态方法：整个静态方法，做用于全部对象
修饰类：括号括起来的部分，做用于全部类
原子性 - 对比
synchronized：不可中断锁，适合竞争不激烈，可读性好
Lock：可中断锁，多样化同步，竞争激烈时能维持常态
Atomic：竞争激烈时能维持常态，比Lock性能好；只能同步一个值api

2. 可见性：一个线程对主内存的修改能够及时的被其余线程观察到。数组

致使共享变量在线程见不可见的缘由：缓存

  1. 线程交叉执行
  2. 冲排序结合线程交叉执行
  3. 共享变量更新后的值没有在工做内存与主内存之间急事更新

synchronized、volatile
JMM关于synchronized的两条规定：安全

  1. 线程解锁前，必须把共享变量的最新制刷新到主内存
  2. 线程加锁前，将清空工做内存中共享变量的值，从而使用共享变量时须要从主内存中从新读取最新的值（注意：加锁与解锁是同一把锁）

volatile - 经过加入内存屏障和禁止重排序优化来实现多线程

  1. 对volatile变量写操做时，会在写操做后加入一条store屏障指令，将本地内存中的共享变量值刷新到主内存
  2. 对volatile变量读操做时，会在读操做前加入一条load屏障指令，从主内存中读取共享变量
  3. volatile变量在每次被线程访问时，都强迫从主内存中读取该变量的值，而当变量的值发生变化时，又会强迫线程将该变量最新的值强制刷新到主内存，这样一来，任什么时候候不一样的线程总能看到该变量的最新值

3. 有序性：一个线程观察其余线程中的指令执行顺序，因为指令重排序的存在，该观察结果通常杂乱无序。并发

Java内存模型中，容许编译器和处理器对指令进行重排序，可是重排序过程不会影响到单线程程序的执行，却会影响到多线程并发执行的正确性。volatile、synchronized、Lock。
【volatile变量规则】：对一个变量的写操做先行发生于后面对这个变量的读操做。（若是一个线程进行写操做，一个线程进行读操做，那么写操做会先行于读操做。）
【传递规则】：若是操做A先行于操做B，而操做B又先行于操做C，那么操做A就先行于操做C。
【线程启动规则】：Thread对象的start方法先行发生于此线程的每个动做。
【线程中断规则】：对线程interrupt方法的调用先行发生于被中断线程的代码检测到中断事件的发生。
【线程终结规则】：线程中全部的操做都先行发生于线程的终止检测，咱们能够经过Thread.join()方法结束、Thread.isAlive()方法的返回值手段检测到线程已经终止执行。
【对象终结规则】：一个对象的初始化完成先行发生于他的finalize()方法的开始。

2、发布对象

发布对象：使一个对象可以被当前范围以外的代码所用。
对象溢出：一种错误的发布。当一个对象尚未构造完成时，就使它被其余线程所见。

3、安全发布对象

在静态初始化函数中初始化一个对象
将对象的引用保存到volatile类型域或者AtomicReference对象中
将对象的引用保存到某个正确构造对象的final类型域中
将对象的引用保存到一个由锁保护的域中 
/**
 * 懒汉模式
 * 双重同步锁单例模式
 * @author Guo
 *
 */
public class SingletonExample1 {
    
    private SingletonExample1(){
        
    }
    
    // volatile禁止指令重排
    private volatile static SingletonExample1 instance = null;
    
    public static SingletonExample1 getInstance(){
        if(instance == null){
            synchronized(SingletonExample1.class){
                if(instance == null){
                    instance = new SingletonExample1();
                }
            }
        }
        return instance;
    }

}

4、避免并发两种方式

  1. 不可变对象
  2. 线程封闭

线程封闭: 把对象封装到一个线程里，只有这一个线程能够看到这个对象，即便这个对象不是线程安全也不会出现任何线程安全问题，由于只在一个线程里框架

  1. 堆栈封闭：局部变量，无并发问题。栈封闭是咱们编程当中遇到的最多的线程封闭。什么是栈封闭呢？简单的说就是局部变量。多个线程访问一个方法，此方法中的局部变量都会被拷贝一分儿到线程栈中。因此局部变量是不被多个线程所共享的，也就不会出现并发问题。因此能用局部变量就别用全局的变量，全局变量容易引发并发问题。
  2. ThreadLocal线程封闭：比较推荐的线程封闭方式。
    【ThreadLocal结合filter完成数据保存到ThreadLocal里，线程隔离。】经过filter获取到数据，放入ThreadLocal, 当前线程处理完以后interceptor将当前线程中的信息移除。使用ThreadLocal是实现线程封闭的最好方法。ThreadLocal内部维护了一个Map，Map的key是每一个线程的名称，而Map的值就是咱们要封闭的对象。每一个线程中的对象都对应着Map中一个值，也就是ThreadLocal利用Map实现了对象的线程封闭

5、线程不安全类与写法

【线程不安全】：若是一个类类对象同时能够被多个线程访问，若是没有作同步或者特殊处理就会出现异常或者逻辑处理错误。
【1. 字符串拼接】：
StringBuilder（线程不安全）、
StringBuffer（线程安全）
【2. 日期转换】:
SimpleDateFormat（线程不安全,最好使用局部变量[堆栈封闭]保证线程安全）
JodaTime 推荐使用（线程安全）
【3. ArrayList、HashSet、HashMap等Collections】:
ArrayList（线程不安全）
HashSet（线程不安全）
HashMap（线程不安全）
【**同步容器**synchronized修饰】
Vector、Stack、HashTable
Collections.synchronizedXXX（List、Set、Map）
【**并发容器** J.U.C】
ArrayList -> CopyOnWriteArrayList：（读时不加锁，写时加锁，避免复制多个副本出来将数据搞乱）写操做时复制，当有新元素添加到CopyOnWriteArrayList中时，先从原有的数组中拷贝一份出来，在新的数组上进行写操做，写完以后再将原来的数组指向新的数组。

clipboard.png

HashSet、TreeSet -> CopyOnWriteArraySet、ConcurrentSkipListSet：
HashMap、TreeMap -> ConcurrentHashMap、ConcurrentSkipListMap：
相比ConcurrentHashMap，ConcurrentSkipListMap具备以下优点：jvm

  1. ConcurrentSkipListMap的存取速度是ConcurrentSkipListMap的4倍左右
  2. ConcurrentSkipListMap的key是有序的
  3. ConcurrentSkipListMap支持更高的并发（它的存取时间和线程数几乎没有关系，更高并发的场景下越能体现出优点）

6、安全共享对象策略 - 总结

  1. 线程限制：一个被线程限制的对象，由线程独占，而且只能被占有它的线程修改
  2. 共享只读：一个共享只读的对象，在没有额外同步的状况下，能够被多个线程并发访问，可是任何线程都不能修改它
  3. 线程安全对象：一个线程安全的对象或者容器，在内部经过同步机制来保证线程安全，因此其余线程无需额外的同步就能够经过公共接口随意访问它
  4. 被守护对象：被守护对象只能经过获取特定锁来访问

7、J.U.C 之 AQS

7.一、 AQS

AQS：AbstractQueneSynchronizer

  1. 使用Node实现FIFO队列，能够用于构建锁或者其余同步装置的基础框架
  2. 利用int类型表示状态
  3. 使用方法是继承
  4. 子类经过继承并经过实现它的方法管理其状态{ acquire和release }的方法操纵状态
  5. 能够同时实现排它锁和共享锁模式（独占、共享）

7.二、 AQS的同步组件以下:

7.2.一、CountDownLatch：闭锁，经过计数来保证线程是否一直阻塞.
CountDownLatch是经过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了本身的任务后，计数器的值就会减1。当计数器值到达0时，它表示全部的线程已经完成了任务，而后在闭锁上等待的线程就能够恢复执行任务。构造器中的计数值（count）实际上就是闭锁须要等待的线程数量。这个值只能被设置一次，并且CountDownLatch没有提供任何机制去从新设置这个计数值。

与CountDownLatch的第一次交互是主线程等待其余线程。主线程必须在启动其余线程后当即调用CountDownLatch.await()方法。这样主线程的操做就会在这个方法上阻塞，直到其余线程完成各自的任务。

其余N 个线程必须引用闭锁对象，由于他们须要通知CountDownLatch对象，他们已经完成了各自的任务。这种通知机制是经过 CountDownLatch.countDown()方法来完成的；每调用一次这个方法，在构造函数中初始化的count值就减1。因此当N个线程都调 用了这个方法，count的值等于0，而后主线程就能经过await()方法，恢复执行本身的任务。

解释一下CountDownLatch概念?
`CountDownLatch`和 `CyclicBarrier`的不一样之处?
给出一些CountDownLatch使用的例子?
 CountDownLatch类中主要的方法?

clipboard.png

public class CountDownLatchExample1 {
    
    // 线程数
    private final static int threadCount = 200;
    
    public static void main(String[] args) throws InterruptedException{
        // 使用线程池进行调度
        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    System.out.println("exception:" + e);
                }finally{
                    countDownLatch.countDown(); // 计数器减一
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MILLISECONDS);
        System.out.println("===finished===");
        exec.shutdown();
    }
    
    
    private static void test(int threadNum) throws InterruptedException{
        Thread.sleep(100);
        System.out.println("threadNum:" + threadNum);
    }
    

}
7.2.二、 Semaphore（信号量）：能够控制同一时间并发线程的数目 
主要函数：acquire、release、tryAcquire
public class SemaphoreExample1 {
    
    // 线程数
    private final static int threadCount = 20;
    
    public static void main(String[] args) throws InterruptedException{
        // 使用线程池进行调度
        ExecutorService exec = Executors.newCachedThreadPool();
        //并发控制(容许并发数20)
        final Semaphore semaphore = new Semaphore(3);
        
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    if(semaphore.tryAcquire(5, TimeUnit.SECONDS)){
                        test(threadNum);
                        semaphore.release();
                    }
                    /** 多个许可：在代码中一共有10个许可，每次执行semaphore.acquire（5）；
                     * 代码时耗费掉5个，因此20/5=4，
                     * 说明同一时间只有4个线程容许执行acquire（）和release（）之间的代码。
                     * */
//                    semaphore.acquire(3); // 获取许可
//                    test(threadNum);
//                    semaphore.release(3); // 释放许可
                } catch (Exception e) {
                    System.out.println("exception:" + e);
                }finally{
                    countDownLatch.countDown(); // 计数器减一
                }
            });
        }
//        countDownLatch.await(100, TimeUnit.MILLISECONDS);
        System.out.println("===finished===");
        exec.shutdown();
    }
    
    
    private static void test(int threadNum) throws InterruptedException{
        System.out.println("threadNum:" + threadNum);
        Thread.sleep(1000);
    }
    

}
7.2.三、 CyclicBarrier：能够完成多个线程之间相互等待，只有当每一个线程都准备就绪后，才能各自继续往下执行 
应用场景：须要全部的子任务都完成时，才执行主任务，这个时候就能够选择使用CyclicBarrier。

简单理解【`人满发车`】：
长途汽车站提供长途客运服务。
当等待坐车的乘客到达20人时，汽车站就会发出一辆长途汽车，让这20个乘客上车走人。
等到下次等待的乘客又到达20人是，汽车站就会又发出一辆长途汽车。
public class CyclicBarrierExample1 {
    
    // 线程数
    private final static int threadCount = 10;
    
    // 屏障的线程数目 5 
    private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
        System.out.println("===continue===");
    });
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool(); 
        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            Thread.sleep(500);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        
        
    }

    private static void race(int threadNum) throws Exception {
        Thread.sleep(1000);
        System.out.println("===" + threadNum + " is ready.");
        try{
            barrier.await(2000, TimeUnit.MILLISECONDS);
        }catch(Exception e){
            System.out.println("e:"+e);
        }
        System.out.println("===" + threadNum + " continue");
    }

}
7.2.四、ReentrantLock 
1. api:
    - lock()
    - unlock()
    - tryLock()
private static Lock lock = new ReentrantLock();
   private static void test(int threadNum){
           lock.lock();
           try{
               count++;
           }finally{
               lock.unlock();
           }
       }
2. ReentrantLock和synchronized的区别
    - 1. `可重入性`
    - 2. `锁的实现`：synchronized是jvm实现，ReentrantLock是jdk实现
    - 3. `性能区别`
    - 4. `功能方面的区别`
3. ReentrantLock独有的功能
    - 1. 可指定是公平锁仍是非公平锁，synchronized只能是非公平锁（公平锁：先等待的线程先得到锁）
    - 2. 提供了一个Condition类，能够分组唤醒须要唤醒的线程
    - 3. 提供可以中断等待锁的线程的机制，lock.lockInterruptibly()
4. ReentrantReadWriteLock
5. StampedLock
6. 锁的使用
   - 当只有少许竞争者线程的时候，`synchronized`是一个很好的通用的锁的实现（synchronized不会引起死锁，jvm会自动解锁）
   - 竞争者线程很多，可是线程增加的趋势是能够预估的，这时候使用`ReentrantLock`是一个很好的通用的锁的实现
7.2.五、Condition 
public class LockExample3 {
public static void main(String[] args){
    ReentrantLock reentrantLock = new ReentrantLock();
    Condition condition = reentrantLock.newCondition();
    int u=1;
    
    
    new Thread(() -> {
        try{
            reentrantLock.lock();
            System.out.println("wait signal"); // 1
            condition.await();
        }catch(InterruptedException e){
            e.printStackTrace();
        }
        System.out.println("get signal");
        reentrantLock.unlock();
    }).start();
    
    new Thread(() -> {
        reentrantLock.lock();
        System.out.println("get lock");
        try{
            Thread.sleep(3000);
        }catch(InterruptedException e){
            e.printStackTrace();
        }
        condition.signalAll();
        System.out.println("send signal");
        reentrantLock.unlock();
    }).start();
    
    
}

}
7.2.六、FutureTask 
建立线程两种方式继承Thread，实现Runnable接口，这两种方式，在任务执行完毕以后获取不到执行结果
    FutureTask、Callable能够获取到执行结果
    1. Callable和Runnable对比
    2. Future接口
    3. FutureTask
    ```
    public static void main(String[] args) throws InterruptedException, ExecutionException {
    FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
        @Override
        public String call() throws Exception {
            System.out.println("do something in callable...");
            Thread.sleep(3000);
            return "Done";
        }
    });
    
    new Thread(futureTask).start();
    System.out.println("do something in main...");
    Thread.sleep(1000);
    String result = futureTask.get();
    System.out.println("result:"+result);
}

}

7.2.七、Fork/Join框架：将大模块切分红多个小模块进行计算

8、线程池

初始化好线程池实例以后，将任务丢进去等待调度执行。

8.一、Thread弊端

  1. 每次new Thread都要新建对象，性能差
  2. 线程缺少统一管理，可能无限制的新建线程，相互竞争，有可能占用过多的系统资源致使死机或者OOM
  3. 缺乏更多功能，如更多执行，按期执行，线程中断

8.二、线程池的好处

  1. 能够重用存在的线程，减小对象的建立、消亡的开销，性能佳
  2. 能够有效的控制最大并发数，提供系统资源利用率，同时能够避免过多的资源竞争，避免阻塞
  3. 提供定时执行、按期执行、单线程、并发数控制等功能
  4. 【ThreadPoolExecutor的初始化参数】
    corePoolSize：核心线程数量
    maximumPoolSize：县城最大线程数
    workQueue：阻塞队列，存储等待执行的任务，很重要，会对线程池运行过程产生重大影响
    keepAliveTime：线程没有任务执行时，最多保持多久时间终止
    unit：keepAliveTime的时间单位
    hreadFactory：线程工厂，用来建立线程
    rejectHandler：当拒绝处理任务时的策略

线程池-ThreadPoolExecutor状态
clipboard.png

线程池-ThreadPoolExecutor方法

1. execute()：提交任务，交给线程池执行
2. submit()：提交任务可以返回执行结果execute + Future
3. shutdown()：关闭线程池，等待任务都执行完
4. shutdownNow()：关闭线程池，不等待任务执行完
5. getTaskCount()：线程池已执行和未执行的任务总数
6. getCompletedTaskCount()：已完成的任务总数
7. getPoolSize()：线程池当前的线程数量
8. getActiveCount：当前线程池中正在执行任务的线程数量

8.三、线程池 - Executors框架（建立线程池）

  1. Executors.newCachedThreadPool：建立一个可缓存的线程池，若是线程池长度超过了处理的须要能够灵活回收空闲线程，若是没有能够回收的，那么就新建线程
public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        // 往线程池中听任务
        for (int i = 0; i < 10; i++) {
            final int index = i; // 任务的序号
            executorService.execute(() -> {
                System.out.println("===task:"+index);
            });
        }
        executorService.shutdown(); // 关闭线程池
    }
  1. Executors.newFixedThreadPool：建立的是一个定长的线程池，能够控制线程的最大并发数，超出的线程会在队列中等待
  2. Executors.newScheduledThreadPool：建立的也是定长线程池，支持定时以及周期性的任务执行
public static void main(String[] args) {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
        
        // 往线程池中听任务
        executorService.scheduleAtFixedRate(() -> {
            log.info("===sechedule run");
        }, 1, 3, TimeUnit.SECONDS); // 延迟一秒，每隔三秒执行任务
        
        
        executorService.schedule(() -> {
            log.info("===sechedule run");
        }, 3, TimeUnit.SECONDS);
        
        executorService.shutdown(); // 关闭线程池
    }
  1. Executors.newSingleThreadExecutor：建立的是一个单线程化的线程池，会用惟一的一个工做线程来执行任务，保证全部任务按照指令顺序执行（指令顺序能够指定它是按照先入先出，优先级执行）

newSingleThreadExecutor打印结果是按照顺序输出
clipboard.png

8.四、线程池 - 合理配置

1. CPU密集型任务，就须要尽可能压榨CPU，参考能够设置为NCPU+1
2. IO密集型任务，参考能够设置为2*NCPU
> NCPU = CPU的数量
> UCPU = 指望对CPU的使用率 0 ≤ UCPU ≤ 1
> W/C = 等待时间与计算时间的比率
> 若是但愿处理器达到理想的使用率，那么线程池的最优大小为：
> 线程池大小=NCPU *UCPU(1+W/C)

https://www.cnblogs.com/super...
https://www.cnblogs.com/super...

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程