Java线程学习入门（三）

开始时间：2022-09-11

公平锁和非公平锁

非公平锁:线程饿死
效率高
公平锁:阳光普照（先进先出）
效率相对低

非公平锁可能会出现线程饿死的情况
他自己抢占到时间片后，一口气就执行完了，另外的线程就饿死了

private final ReentrantLock lock = new ReentrantLock(true);
1

可重入锁

可重入锁又叫递归锁

package com.bupt.syn;

public class SyncLockDemo {
    public static void main(String[] args) {
        Object object = new Object();
        new Thread(() -> {
            synchronized (object) {
                System.out.println(Thread.currentThread().getName() + "最外层");
                synchronized (object) {
                    System.out.println(Thread.currentThread().getName() + "中层");
                    synchronized (object) {
                        System.out.println(Thread.currentThread().getName() + "最内层");
                    }
                }
            }
        }, "AA").start();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

我们来看这个例子，如果synchronized不是可重入锁
那么我在第一次锁住object后，想要进入中层，应该是办不到的，因为此时外层的synchronize同步语句块并没有执行结束，按理说他的锁并没有释放出来，怎么又能获得呢？
但因为他是可重入锁，所以可以再次获取到

AA最外层
AA中层
AA最内层
1
2
3

同理可以知道Reentranlock也是可重入锁

        new Thread(() -> {
            try {
                lock.lock();
                System.out.println(Thread.currentThread().getName() + "最外层");
                try {
                    lock.lock();
                    System.out.println(Thread.currentThread().getName() + "中间层");
                } finally {
                    lock.unlock();
                }
            } finally {
                lock.unlock();
            }
        }, "AA").start();
1
2
3
4
5
6
7
8
9
10
11
12
13
14

AA最外层
AA中间层
1
2

死锁

package com.bupt.syn;

public class DeadLock {
    public static void main(String[] args) {
        DeadLockDemo A = new DeadLockDemo(true);
        A.setName("A线程");
        DeadLockDemo B = new DeadLockDemo(false);
        B.setName("B线程");
        A.start();
        B.start();

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

class DeadLockDemo extends Thread {
    static Object o1 = new Object();
    static Object o2 = new Object();
    boolean flag;

    public DeadLockDemo(boolean flag) {
        this.flag = flag;
    }

    @Override
    public void run() {
        if (flag) {
            synchronized (o1) {
                System.out.println(Thread.currentThread().getName() + "进入1");
                synchronized (o2) {
                    System.out.println(Thread.currentThread().getName() + "进入2");
                }
            }
        } else {
            synchronized (o2) {
                System.out.println(Thread.currentThread().getName() + "进入3");
                synchronized (o1) {
                    System.out.println(Thread.currentThread().getName() + "进入4");
                }
            }
        }
    }
}

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

然后一直卡住

A线程进入1
B线程进入3
1
2

面试如果手写死锁，就可以这样写

package com.bupt.syn;

public class DeadLock {
    static Object a=new Object();
    static Object b=new Object();
    public static void main(String[] args) {
        new Thread(()->{
            synchronized (a){
                System.out.println("线程"+Thread.currentThread().getName()+"持有锁A");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (b){
                    System.out.println("线程"+Thread.currentThread().getName()+"持有锁B");
                }
            }
        },"AA").start();
        new Thread(()->{
            synchronized (b){
                System.out.println("线程"+Thread.currentThread().getName()+"持有锁B");
                synchronized (a){
                    System.out.println("线程"+Thread.currentThread().getName()+"持有锁A");
                }
            }
        },"BB").start();

    }
}


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

验证是否为死锁
使用命令jps

F:\编程学习\多线程>jps
10656
5072 DeadLock
16792 Launcher
15228 Jps
1
2
3
4
5

实现Callable类

依然先看JavaGuide

Runnable 接口 不会返回结果或抛出检查异常，但是 Callable 接口 可以。所以，如果任务不需要返回结果或抛出异常推荐使用 Runnable 接口 ，这样代码看起来会更加简洁。

前者用run，后者用call

class MyThread1 implements Runnable {
    @Override
    public void run() {
        System.out.println("测试Runnable");
    }
}

class MyThread2 implements Callable {

    @Override
    public Integer call() throws Exception {
        System.out.println("测试Callable");
        return 100;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

我们观测到没办法直接作为参数传递进去
那么就要找既Callable和Runnable的桥梁

FutureTask

public static void main(String[] args) throws ExecutionException, InterruptedException {
        //使用lambda表达式类实现Callable
        FutureTask<Integer> futureTask2 = new FutureTask<>(() -> {
            System.out.println(Thread.currentThread().getName() + " come in Callable");
            //Callable返回值
            return 1024;
        });
        new Thread(futureTask2, "lucy").start();
        while (!futureTask2.isDone()) {
            System.out.println("wait...");
        }
        //第一次要计算，计算会等待
        System.out.println("第一次拿到结果:" + futureTask2.get());
        //第二次直接返回结果
        System.out.println("第二次拿到结果:" + futureTask2.get());
        System.out.println(Thread.currentThread().getName() + " over");
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

我们Callable返回值就是1024
通过使用lambda表达式，利用中间商FutureTask来得到对象
再把futureTask放入Thread里面
通过futureTask得到Callable的值

AQS组件

CountDownLatch

示例：六个同学全部离开，班长才能锁门

public static void main(String[] args) throws InterruptedException {
        for (int i = 1; i <= 6; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "号同学离开了教室");
            }, String.valueOf(i)).start();
        }
        System.out.println(Thread.currentThread().getName() + "班长锁门");
    }
1
2
3
4
5
6
7
8

我们看看输出

2号同学离开了教室
5号同学离开了教室
6号同学离开了教室
4号同学离开了教室
3号同学离开了教室
main班长锁门
1号同学离开了教室
1
2
3
4
5
6
7

main先锁门了

解决方案

public static void main(String[] args) throws InterruptedException {
        CountDownLatch count = new CountDownLatch(6);
        for (int i = 1; i <= 6; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "号同学离开了教室");
                count.countDown();
            }, String.valueOf(i)).start();
        }
        //等待
        count.await();
        System.out.println(Thread.currentThread().getName() + "班长锁门");
    }
1
2
3
4
5
6
7
8
9
10
11
12

3号同学离开了教室
6号同学离开了教室
5号同学离开了教室
1号同学离开了教室
4号同学离开了教室
2号同学离开了教室
main班长锁门
1
2
3
4
5
6
7

CyclicBarrier

集齐7颗龙珠召唤神龙

来看这一个Demo

package com.bupt.juc;

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

public class CyclicBarrierDemo {
    private static final int NUMBER = 7;

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(NUMBER, () -> {
            System.out.println("集齐7颗龙珠就可以召唤神龙");
        });
        for (int i = 1; i <=7; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "星龙珠");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                } finally {
                }
            }, String.valueOf(i)).start();
        }
    }
}
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

设定一个目标值，每调用一个线程，如果总的线程数没达到目标值，就await

2星龙珠
7星龙珠
6星龙珠
4星龙珠
5星龙珠
3星龙珠
1星龙珠
集齐7颗龙珠就可以召唤神龙
1
2
3
4
5
6
7
8

Semophore

六辆车，三个停车位
通过信号量的PV操作
先设置总资源数3
再轮着来acquire和release

package com.bupt.juc;

import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 1; i <= 6; i++) {
            new Thread(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + " 抢到了车位");
                    TimeUnit.SECONDS.sleep(new Random().nextInt(5));
                    System.out.println(Thread.currentThread().getName() + " 离开了车位");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();
                }
            }, String.valueOf(i)).start();
        }
    }
}
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

1 抢到了车位
3 抢到了车位
2 抢到了车位
3 离开了车位
6 抢到了车位
1 离开了车位
5 抢到了车位
6 离开了车位
4 抢到了车位
4 离开了车位
5 离开了车位
2 离开了车位
1
2
3
4
5
6
7
8
9
10
11
12

读写锁

读写锁:一个资源可以被多个读线程访问，或者可以被一个写线程访问，但是不能同时存在读写线程，读写互斥，读读共享的

表锁，锁一整张表
行锁，锁一行
行锁容易发生死锁

读锁：共享锁，发生死锁
写锁：独占锁，也可能发生死锁
不加读写锁时

package com.bupt.readWrite;

import java.util.HashMap;

public class ReadWriteLockDemo {
    public static void main(String[] args) {
        MyCache myCache = new MyCache();
        for (int i = 1; i <= 5; i++) {
            final int num = 1;
            new Thread(() -> {
            //这个传参只能传被final修饰的
                myCache.put(num + "", num + "");
            }, String.valueOf(i)).start();
        }
        for (int i = 1; i <= 5; i++) {
            final int num = 1;
            new Thread(() -> {
                myCache.get(num + "");
            }, String.valueOf(i)).start();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

class MyCache {
    private volatile HashMap<String, Object> map = new HashMap<>();

    public void put(String key, Object value) {
        System.out.println(Thread.currentThread().getName() + "正在写操作" + key);
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        map.put(key, value);
        System.out.println(Thread.currentThread().getName() + "写完了");
    }

    public Object get(String key) {
        System.out.println(Thread.currentThread().getName() + "正在读取操作" + key);
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Object object = map.get(key);
        System.out.println(Thread.currentThread().getName() + "读完了");
        return object;
    }
}
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

查看一下输出结果

2正在写操作1
5正在写操作1
4正在写操作1
3正在写操作1
1正在写操作1
1正在读取操作1
2正在读取操作1
3正在读取操作1
4正在读取操作1
5正在读取操作1
2写完了
2读完了
5写完了
4读完了
4写完了
3写完了
1写完了
1读完了
5读完了
3读完了

Process finished with exit code 0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

我们可以观测到，我的线程在没有写完数据时，该线程已经在读了，这肯定是不对的
我们需要正在写->写完了->正在读->读完了这个顺序
重新改进一下

class MyCache {
    private volatile HashMap<String, Object> map = new HashMap<>();
    //创建读写锁对象
    private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    public void put(String key, Object value) {
        //添加写锁
        readWriteLock.writeLock().lock();

        try {
            System.out.println(Thread.currentThread().getName() + "正在写操作" + key);
            Thread.sleep(300);
            map.put(key, value);
            System.out.println(Thread.currentThread().getName() + "写完了" + key);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            readWriteLock.writeLock().unlock();
        }
    }

    public Object get(String key) {
        readWriteLock.readLock().lock();
        Object object = null;
        try {
            System.out.println(Thread.currentThread().getName() + "正在读取操作" + key);
            Thread.sleep(300);
            object = map.get(key);
            System.out.println(Thread.currentThread().getName() + "读完了");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
        return object;
    }
}
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

此时就是按顺序执行了
此时可以看到写的时候是一个线程走完
读的时候可以一起来读

4正在写操作1
4写完了1
5正在写操作1
5写完了1
3正在写操作1
3写完了1
2正在写操作1
2写完了1
1正在写操作1
1写完了1
1正在读取操作1
2正在读取操作1
3正在读取操作1
4正在读取操作1
5正在读取操作1
1读完了
4读完了
2读完了
5读完了
3读完了
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

读读可以共享，提升性能
同时多人进行读操作
写写的时候只能一个线程一个线程来
读的时候不能写，只有读完成后才能写

写的时候，同一个线程可以读，跨线程的不能读
不同线程之间，读写互斥，同一线程可以先获取写锁，再获取读锁，反过来不行

锁降级

我测试了一下

//可重入读写锁对象
        ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
        ReentrantReadWriteLock.ReadLock readLock = rwLock.readLock();
        ReentrantReadWriteLock.WriteLock writeLock = rwLock.writeLock();
1
2
3
4

//锁降级
        //获取写锁
        writeLock.lock();
        System.out.println("jindaohei");
        //获取读锁
        readLock.lock();
        System.out.println("--read");
        //释放写锁
        writeLock.unlock();
        //释放读锁
        readLock.unlock();
1
2
3
4
5
6
7
8
9
10
11

能正常执行并结束

jindaohei
--read
1
2

写锁获取后还能获取读锁
但是反过来不行

//获取读锁
        readLock.lock();
        System.out.println("--read");
        //获取写锁
        writeLock.lock();
        System.out.println("jindaohei");

        //释放读锁
        readLock.unlock();
        //释放写锁
        writeLock.unlock();
1
2
3
4
5
6
7
8
9
10
11

一直卡死

--read
1

读完释放后才能写

阻塞队列

Blocking Queue

支持两个附加操作的队列
支持阻塞的插入和移除
生产者添加元素，消费者获取元素，阻塞队列就是生产者存放元素，消费者获取元素的容器

用add添加，如果超过容量，报异常，如果是offer添加超过容量，返回false

public class QueueDemo01 {
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
        //第一组
        blockingQueue.offer("a");
        blockingQueue.offer("b");
        blockingQueue.offer("c");
        System.out.println(blockingQueue.offer("d"));
        System.out.println(blockingQueue.poll());
    }
}
1
2
3
4
5
6
7
8
9
10
11

而如果是用put，超标后会一直阻塞，程序不会继续运行
如果是offer，也可以设置超时时间，超过了就退出阻塞

System.out.println(blockingQueue.offer("d",3, TimeUnit.SECONDS));
1

线程池

线程池（英语: thread pool ) :一种线程使用模式。

• 线程过多会带来调度开销，进而影响缓存局部性和整体性能。
• 而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。
• 线程池不仅能够保证内核的充分利用，还能防止过分调度。
  好处
• 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
• 提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
• 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

我们测试一下
当我们使用threadPool1时
不开启sleep

public class ThreadPoolDemo1 {
    public static void main(String[] args) {
        //一个银行，五个窗口
        ExecutorService threadPool1 = Executors.newFixedThreadPool(5);
        //一个银行，一个窗口
        ExecutorService threadPool2 = Executors.newSingleThreadExecutor();
        //处理10个顾客
        try {
            for (int i = 1; i <= 50; i++) {
                threadPool1.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "办理业务");
                });
                //Thread.sleep(1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool1.shutdown();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

pool-1-thread-3办理业务
pool-1-thread-5办理业务
pool-1-thread-4办理业务
pool-1-thread-4办理业务
pool-1-thread-4办理业务
pool-1-thread-1办理业务
pool-1-thread-2办理业务
pool-1-thread-2办理业务
pool-1-thread-2办理业务
pool-1-thread-1办理业务
pool-1-thread-4办理业务
pool-1-thread-4办理业务
......
1
2
3
4
5
6
7
8
9
10
11
12
13

5个Thread滚动使用，可以插着使用
我加了一个sleep后，那就是按1 2 3 4 5 来走的

而使用threadPool2
则只靠一个池子干完所有事情

而我如果使用

ExecutorService threadPool3 = Executors.newCachedThreadPool();
1

他根据需要自己定义要创建多少个线程，处理100个请求这把最多开到19个线程
而我处理一百个，我还在上面加一个休眠时间，此时他一个线程就干完了我100个请求

这三个类的底层都是ThreadPoolExecutor

ThreadPoolExecutor

7个核心参数

执行了execute，线程才创建

我们自己来测试一下这个

《阿里巴巴 Java 开发手册》中强制线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险

Executors 返回线程池对象的弊端如下：

FixedThreadPool 和 SingleThreadExecutor ： 允许请求的队列长度为 Integer.MAX_VALUE> ，可能堆积大量的请求，从而导致 OOM。 CachedThreadPool 和 ScheduledThreadPool ：> 允许创建的线程数量为 Integer.MAX_VALUE ，可能会创建大量线程，从而导致 OOM。

package com.bupt.pool;

import com.bupt.javaGuide.MyRunnable;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolDemo2 {
    private static final int CORE_POOL_SIZE = 5;
    private static final int MAX_POOL_SIZE = 10;
    private static final int QUEUE_CAPACITY = 10;
    private static final Long KEEP_ALIVE_TIME = 1L;
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy());

        for (int i = 0; i < 20; i++) {
            //创建WorkerThread对象（WorkerThread类实现了Runnable 接口）
            Runnable worker = new MyRunnable("" + i);
            //执行Runnable
            executor.execute(worker);
        }
        //终止线程池
        executor.shutdown();
        while (!executor.isTerminated()) {
        }
        System.out.println("Finished all threads");
    }
}

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

比较一下
核心线程数5，队列10，最大线程数10
工作线程个数为10，开五个线程，五个干完再去拿剩下五个
工作线程个数为15，开五个线程，五个干完拿五个，再拿五个
工作线程个数为16，开六个线程，六个干完再去拿
工作线程个数为17，开七个线程。。
工作线程个数为19，开九个线程

另外要注意的是，比如我开了16个线程，按理说先处理的1-5 6-15在排队，但此时来了16，新开工作线程其实执行的是16，而不是从队列里拿

拒绝策略

ThreadPoolExecutor.AbortPolicy ：抛出 RejectedExecutionException来拒绝新任务的处理。
ThreadPoolExecutor.CallerRunsPolicy ：调用执行自己的线程运行任务，也就是直接在调用execute方法的线程中运行(run)被拒绝的任务，如果执行程序已关闭，则会丢弃该任务。因此这种策略会降低对于新任务提交速度，影响程序的整体性能。如果您的应用程序可以承受此延迟并且你要求任何一个任务请求都要被执行的话，你可以选择这个策略。
ThreadPoolExecutor.DiscardPolicy ：不处理新任务，直接丢弃掉。
ThreadPoolExecutor.DiscardOldestPolicy ： 此策略将丢弃最早的未处理的任务请求。

测试一下

 new ThreadPoolExecutor.AbortPolicy());
1

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task 20 rejected from java.util.concurrent
1

为什么要先塞满队列再开新线程

因为开线程的开销比较大，直接从队列里拿任务执行比较方便

fork和join

拆分再合并，
Fork是拆，Join是合并
分治法

package com.bupt.forkjoin;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class ForkJoinDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyTask myTask = new MyTask(0, 100);
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(myTask);
        Integer result = forkJoinTask.get();
        System.out.println(result);
        forkJoinPool.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

class MyTask extends RecursiveTask<Integer> {
    //拆分差值不能超过10
    private static final Integer VALUE = 10;
    private int begin;
    private int end;
    private int result;

    public MyTask() {
    }

    public MyTask(int begin, int end) {
        this.begin = begin;
        this.end = end;
    }

    //拆分和合并
    @Override
    protected Integer compute() {
        //判断相加的两个数是否大于10
        if (end - begin <= VALUE) {
            //相加
            for (int i = begin; i <= end; i++) {
                result += i;
            }
        } else {
            int middle = begin + (end - begin) / 2;
            MyTask myTask01 = new MyTask(begin, middle);
            MyTask myTask02 = new MyTask(middle + 1, end);
            myTask01.fork();
            myTask02.fork();
            result = myTask01.join() + myTask02.join();
        }
        return result;
    }
}

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

结束时间：2022-09-11