Java多线程&高并发

摘要：线程启动规则对象的方法先行发生于此线程的每一个动作。所以局部变量是不被多个线程所共享的，也就不会出现并发问题。通过获取到数据，放入当前线程处理完之后将当前线程中的信息移除。主线程必须在启动其他线程后立即调用方法。

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

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

Atomic包：

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

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

2. 可见性：一个线程对主内存的修改可以及时的被其他线程观察到。

导致共享变量在线程见不可见的原因：

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

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

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

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

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

3. 有序性：一个线程观察其他线程中的指令执行顺序，由于指令重排序的存在，该观察结果一般杂乱无序。

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

发布对象：使一个对象能够被当前范围之外的代码所用。
对象溢出：一种错误的发布。当一个对象还没有构造完成时，就使它被其他线程所见。

在静态初始化函数中初始化一个对象
将对象的引用保存到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;
    }

}

不可变对象
线程封闭

线程封闭: 把对象封装到一个线程里，只有这一个线程可以看到这个对象，即使这个对象不是线程安全也不会出现任何线程安全问题，因为只在一个线程里


堆栈封闭：局部变量，无并发问题。栈封闭是我们编程当中遇到的最多的线程封闭。什么是栈封闭呢？简单的说就是局部变量。多个线程访问一个方法，此方法中的局部变量都会被拷贝一分儿到线程栈中。所以局部变量是不被多个线程所共享的，也就不会出现并发问题。所以能用局部变量就别用全局的变量，全局变量容易引起并发问题。

ThreadLocal线程封闭：比较推荐的线程封闭方式。
【ThreadLocal结合filter完成数据保存到ThreadLocal里，线程隔离。】通过filter获取到数据，放入ThreadLocal, 当前线程处理完之后interceptor将当前线程中的信息移除。使用ThreadLocal是实现线程封闭的最好方法。ThreadLocal内部维护了一个Map，Map的key是每个线程的名称，而Map的值就是我们要封闭的对象。每个线程中的对象都对应着Map中一个值，也就是ThreadLocal利用Map实现了对象的线程封闭

【线程不安全】：如果一个类类对象同时可以被多个线程访问，如果没有做同步或者特殊处理就会出现异常或者逻辑处理错误。
【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中时，先从原有的数组中拷贝一份出来，在新的数组上进行写操作，写完之后再将原来的数组指向新的数组。

HashSet、TreeSet -> CopyOnWriteArraySet、ConcurrentSkipListSet：
HashMap、TreeMap -> ConcurrentHashMap、ConcurrentSkipListMap：
相比ConcurrentHashMap，ConcurrentSkipListMap具有如下优势：

ConcurrentSkipListMap的存取速度是ConcurrentSkipListMap的4倍左右

ConcurrentSkipListMap的key是有序的

ConcurrentSkipListMap支持更高的并发（它的存取时间和线程数几乎没有关系，更高并发的场景下越能体现出优势）

线程限制：一个被线程限制的对象，由线程独占，并且只能被占有它的线程修改

共享只读：一个共享只读的对象，在没有额外同步的情况下，可以被多个线程并发访问，但是任何线程都不能修改它

线程安全对象：一个线程安全的对象或者容器，在内部通过同步机制来保证线程安全，所以其他线程无需额外的同步就可以通过公共接口随意访问它

被守护对象：被守护对象只能通过获取特定锁来访问

AQS：AbstractQueneSynchronizer

使用Node实现FIFO队列，可以用于构建锁或者其他同步装置的基础框架
利用int类型表示状态
使用方法是继承

子类通过继承并通过实现它的方法管理其状态{ acquire和release }的方法操纵状态
可以同时实现排它锁和共享锁模式（独占、共享）

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

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

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

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

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.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.3、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.4、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.5、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.6、FutureTask

    创建线程两种方式继承Thread，实现Runnable接口，这两种方式，在任务执行完毕之后获取不到执行结果
    FutureTask、Callable可以获取到执行结果
    1. Callable和Runnable对比
    2. Future接口
    3. FutureTask
    ```
    public static void main(String[] args) throws InterruptedException, ExecutionException {
    FutureTask futureTask = new FutureTask(new Callable() {
        @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.7、Fork/Join框架：将大模块切分成多个小模块进行计算

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

每次new Thread都要新建对象，性能差

线程缺乏统一管理，可能无限制的新建线程，相互竞争，有可能占用过多的系统资源导致死机或者OOM

缺少更多功能，如更多执行，定期执行，线程中断

可以重用存在的线程，减少对象的创建、消亡的开销，性能佳

可以有效的控制最大并发数，提供系统资源利用率，同时可以避免过多的资源竞争，避免阻塞

提供定时执行、定期执行、单线程、并发数控制等功能

【ThreadPoolExecutor的初始化参数】
corePoolSize：核心线程数量
maximumPoolSize：县城最大线程数
workQueue：阻塞队列，存储等待执行的任务，很重要，会对线程池运行过程产生重大影响
keepAliveTime：线程没有任务执行时，最多保持多久时间终止
unit：keepAliveTime的时间单位
hreadFactory：线程工厂，用来创建线程
rejectHandler：当拒绝处理任务时的策略

线程池-ThreadPoolExecutor状态

线程池-ThreadPoolExecutor方法

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

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(); // 关闭线程池
    }

Executors.newFixedThreadPool：创建的是一个定长的线程池，可以控制线程的最大并发数，超出的线程会在队列中等待

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(); // 关闭线程池
    }

Executors.newSingleThreadExecutor：创建的是一个单线程化的线程池，会用唯一的一个工作线程来执行任务，保证所有任务按照指令顺序执行（指令顺序可以指定它是按照先入先出，优先级执行）

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

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...