java多线程

线程状态

NEW：线程创建，还未启动
RUNNABLE：又分ready和running，由线程调度器可以安排执行
WAITING：等待被唤醒
TIMED_WAITING：隔一段时间自动唤醒
BLOCKED：阻塞，只有synchronized才会出现(涉及操作系统层面)
TERMINAL：结束

join()：等待其他线程执行完；线程a中调用线程b.join()：a要等待b执行完才继续执行
如何保证三个线程顺序执行？

1、使用join
2、使用SingleThreadExecutor线程池
3、使用ReentrantLock的condition精确唤醒一个等待的线程
1
2
3

线程中断

interrupt

interrupt：打断某个线程(设置标志位)
isInterrupted：查询是否被打断(查询标志位)
static interrupted：查询当前线程是否被打断，并重置标志位(设为false)；针对的是当前线程

优雅的结束线程：interrupt只是设置标志，结束线程还是要由开发者决定

public static void main(String[] args) {
       Thread t1= new Thread(
           ()->{
               for(;;){
                   if(Thread.currentThread().isInterrupted()){
                       System.out.println("t1："+"tunning");
                       System.out.println("t1："+Thread.currentThread().isInterrupted());
                       break;
                   }
               }
           }
       );
        t1.start();
        try{
            Thread.sleep(5000);
        }catch(Exception e){
        }
        t1.interrupt();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

isInterrupted是针对当前线程

public static void main(String[] args) {
       Thread t1= new Thread(
           ()->{
               for(;;){
                   if(Thread.currentThread().interrupted()){
                       System.out.println("t1："+"tunning");
                       System.out.println("t1："+Thread.currentThread().interrupted());
                       break;
                   }
               }
           }
       );
        t1.start();
        try{
            Thread.sleep(2000);
        }catch(Exception e){
        }
        t1.interrupt();
        Thread.currentThread().interrupt();
        System.out.println("Main thread："+Thread.currentThread().isInterrupted());
        //这里打印的是当前线程，也就是主线程的
        System.out.println("main thread："+t1.interrupted());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

sleep、wait：抛interruptException异常，会自定调用interrupted()方法重置标志位

synchoronized、lock：interrupt对synchoronized和lock方法无效，对lockInterruptibly有效；如果想在获取锁的时候中断线程，可以使用lockInterruptibly结合实际结束线程(interrupt只是设置标志，结束线程还是要由开发者决定)

优雅的结束线程

1、自然结束
2、stop()、suspend()/resume()：粗暴，不保证数据
3、volatile标志：线程阻塞，没办法循环检查标志；打断时间也不是很精确：比如在循环中i++，每次结束i的值都不一样
4、interrupt()/isInterrupt()

并发三大特性

可见性

java线程从主存中读取数据到工作空间，在工作空间修改值后再写到主存中；如果两个线程读取的是同一份数据可能就会出现不一致问题
线程t一直循环，flag被主线程修改了t线程还是使用的自己工作空间的值

static boolean  flag=true;
    public static void main(String []args){
        Thread t=new Thread(()->{
            while(flag){
                System.out.println("running");
            }
            System.out.println("end");
        });
        t.start();
        try{ 
            Thread.sleep(300);
        }catch(Exception e){
        }
        flag=false;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

使用volatile修饰，保证可见性

static volatile boolean  flag=true;
    public static void main(String []args){
        Thread t=new Thread(()->{
            while(flag){
                System.out.println("running");
            }
            System.out.println("end");
        });
        t.start();
        try{ 
            Thread.sleep(300);
        }catch(Exception e){
        }
        flag=false;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

final、synchoronized也可以保证可见性
volatile保证数据一致性：如果读取的变量用volatile修饰，当线程A修改值后，线程B会直接从主存中重新读取最新的值
缓存行：cpu与主存之间并不是直接交互的，中间有L1、L2、L3三层缓存，L3是共享的；
L3存储的是缓存行，一个缓存行大小是64 bytes；
缓存局部性原理：当从主存读取数据x时，相邻的数据可能会被使用，也会一起被读取
如图，如果数据x和y在一个缓存行里，而且左边使用x，右边使用y，即使左边不会使用y，右边不使用x，因为在同一个缓存行里，只要左边修改了x值，就会因为缓存一致性而通知右边重新读取y值；同样右边同理；这样肯定会影响效率，如何解决：在x前后放7个long型变量，y前后放7个long型变量，这样x和y肯定就不会处于同一个缓存行中
缓存一致性协议 ：当缓存中的数据被修改，会通知其他读取相同数据的缓存重新从主存中读取最新的值

有序性

由于cpu指令优化：代码里的语句并不一定是按顺序一条一条执行的，只保证最终一致性；
new 对象分三步：DDL双重校验所需配合volatile
volatile禁止指令重排序

1、分配内存空间，属性赋默认值(半初始化状态)
2、掉构造方法，属性赋值
3、将变量引用与内存关联
1
2
3

volatiler如果修饰的是引用类型，里面的字段还是不能保证可见性；解决办法就是给里面的字段加volatile
volatile
使用内存屏障阻止乱序执行：指令A和指令B中间插入特殊指令(内存屏障)，指令A执行了才能执行B
JVM内存屏障：只是一种协议，不同的JVM需要具体实现；load(读)，store(写)
loadload：指令A(读)->loadload->指令B(读)，不能乱序
storestore：指令A(写)->storestore->指令B(写)，不能乱序
loadstore：指令A(读)->loadstore->指令B(写)，不能乱序
storeload：指令A(写)->storeload->指令B(读)，不能乱序

原子性

在执行时不能被打断
i++：不是原子操作，会被翻译成多条指令执行，执行到某一条时可能会被打断

悲观锁

加锁保证原子性：本质是把并发变为序列化操作(串行)
synchoronized锁升级：无锁->向锁->自旋锁->重量级锁

偏向锁：
	如果只有一个线程：markword记录线程id，如果第二次还是这个线程，直接执行
自旋锁：
	如果有线程竞争升级为自旋锁(cas)
重量级锁：
	自旋10次仍未获得锁，升级为重量级锁(操作系统层面)
1
2
3
4
5
6

自旋锁：占用cpu，但是是在用户态下，cpu指令支持；适用于执行时间短，线程数少
重量级锁：不占用cpu，但是是在内核态下；适用于执行时间长，线程数多
synchoronized锁的是对象而不是代码
synchoronized锁不要用String(字符串常量)、Integer、Long等类型

乐观锁

并发小工具

CAS操作：cup层面实现

reentrantLock

1、lock()，unlock()
2、 tryLock()：尝试获取锁，true：获取到锁，false：没获取到锁
	public boolean tryLock()；
	public boolean tryLock(long timeout,TimeUnit unit) throws InterruptedException
3、lockInterruptibly()
	获取锁的过程可以通过interrupt()设置标志位打断
4、condition：与wait，notify相似，可以精准指定唤醒线程
	public Condition newCondition();
1
2
3
4
5
6
7
8

ReentrantReadWriteLock

readLock() ：读锁，也是共享锁；可以并发读
writeLock() ：写锁，也是排他锁；只能序列写

Semaphore

CyclicBarrier

CountDownLatch

允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助

public static void main(String args[]){
        CountDownLatch start =new CountDownLatch(1);
        CountDownLatch cDownLatch =new CountDownLatch(3);
        new Thread(()->{
            try{
                start.await();
                 System.out.println("玩家1通关");
            }catch(Exception e){

            }finally{
                cDownLatch.countDown();
            }
        }).start();
        new Thread(()->{
            try{
                start.await();
                System.out.println("玩家2通关");
            }catch(Exception e){

            }finally{
                cDownLatch.countDown();
            }
        }).start();
        new Thread(()->{
            try{
                start.await();
                System.out.println("玩家3通关");
            }catch(Exception e){

            }finally{
                cDownLatch.countDown();
            }
        }).start();
        try{
            start.countDown();
            System.out.println("游戏开始");
            cDownLatch.await();
             System.out.println("游戏结束");
        }catch(InterruptedException e){

        }
       
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

运行结果

游戏开始
玩家3通关
玩家2通关
玩家1通关
游戏结束
1
2
3
4
5

exchanger

用于两个线程交换值，是同步阻塞的

lockSupport

阻塞线程：wait()、notify()、notifyall()要在同步代码中调用：因为wait方法就是释放锁，前提就是自身得有锁，notify是唤醒阻塞队列中的线程竞争锁，notify不会释放锁

park() ：阻塞线程
unpark(Thread thread)：线程继续运行

1、unpark可以在park之前调用：因为可能会要些时间才会执行到park
2、park和unpark不需要在同步代码中调用
1
2

ThreadLocal

线程独享

map：threadLocalMap，是Thread类里的；map的key是以threadLocal区分，
两个线程使用threadLocal时：其实使用的是当前线程里的map；
map的key最终是放的entry对象，entry是一个弱引用。

这里如果t1=null，key是强引用的话，threadLocal对象永远都不会被回收。key是弱引用可以回收，但value是强引用，只能调用remove方法回收
thread用完记得调用remove方法

强引用

通常的new对象就是强引用
在引用为null的情况下才会被回收

软引用(SoftReference)

在引用为null的情况下会被回收，在创建新对象内存不足的情况下也会被回收
可以用来做缓存

弱引用(WeakReference)

只要发生gc就可以被回收
threadLocal使用

虚引用(PhantomReference)

只要发生gc就可以被回收
需要和队列关联，当一个有虚引用的对象被回收之前，会把他先放进队列中，开发者监听队列然后做后续操作

java线程池

threadPoolExecutor

线程池执行流程
1

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
1
2
3
4
5
6
7

corePoolSize：核心线程数
1

maximumPoolSize：最大线程数
1

keepAliveTime：空闲线程存活时间
1

unit：时间单位
1

workQueue：阻塞线程队列
1、ArrayBlockingQueue：可指定队列大小
2、LinkedBlockingQueue：队列最大可达Integer.MAX_VALUE
3、SynchronousQueue：队列容量为0，添加线程会阻塞一直到有消费线程消费
4、TransferQueue：可指定容量，添加线程会阻塞一直到有消费线程消费
1
2
3
4
5

hreadFactory：生产线程的工厂
1

handler：拒绝策略，jdk默认提供四种策略
1、 new ThreadPoolExecutor.AbortPolicy()：直接抛异常RejectedExecutionException
2、 new ThreadPoolExecutor.DiscardPolicy()：不抛异常
3、 new ThreadPoolExecutor.DiscardOldestPolicy()：将队列中时间最长的出队，入队新任务
4、 new ThreadPoolExecutor.CallerRunsPolicy()：由调用方线程执行
1
2
3
4
5

默认提供的几种线程池

newFixedThreadPool
1

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
1
2
3
4
5

newSingleThreadExecutor
1

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
1
2
3
4
5
6

newCachedThreadPool：最大线程数：Integer.MAX_VALUE，SynchronousQueue队列容量为0，只要来任务并且没有空闲线程，就会创建线程执行
1

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
1
2
3
4
5

ScheduledThreadPoolExecutor
1

public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }
1
2
3
4
5

Future

针对线程有返回值的，存储线程执行完后的结果

FutureTask

本身既能存储结果，又能作为任务提交
get获取结果是阻塞的：在主线程中调用get会阻塞
…

CompletableFuture

获取返回是在执行线程中，异步的，主线程不会阻塞

	CompletableFuture.supplyAsync(() -> {
                    System.out.println(Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return "hello";
                })
                .whenComplete((s, e) -> {
                    System.out.println(Thread.currentThread().getName() + " : " + s);
                })
                .thenApply(s -> s + " world")
                .thenAccept((s) -> {
                    System.out.println(Thread.currentThread().getName() + " : " + s);
                });
        //pool.shutdown();
        Thread.sleep(3000);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

ForkJoinPool

与threadPoolExecutor不同，每个线程对应一个任务队列

submit()：提交的任务带返回值，任务可以实现RecursiveTask<T>泛型值返回值
1

计算数组求和：

//初始化数组
static long[] arrays=new long[1_00_0000];
for (int i = 0; i < arrays.length; i++) {
            arrays[i]=i;
        }
1
2
3
4
5

任务类

static class Task extends RecursiveTask<Long> {
        private int start;
        private int end;
        long[] arrays;

        public Task(int start, int end, long[] arrays) {
            this.start = start;
            this.end = end;
            this.arrays = arrays;
        }

        @Override
        protected Long compute() {
            int diff = end - start;
            if(diff<100000){
                long sum=0;
                for (int i = start; i <= end; i++) {
                    sum+=arrays[i];
                }
                return sum;
            }else {
            	//开启新任务
                Task task = new Task(start, (end+start) / 2, arrays);
                //开启新任务
                Task task1 = new Task((end+start) / 2+1, end, arrays);
                task.fork();
                task1.fork();
                //结果汇总
                return task.join()+task1.join();
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

提交任务

		ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Long> task = pool.submit(new Task(0, arrays.length - 1, arrays));
        //阻塞get
        System.out.println(task.get());
1
2
3
4