多线程案例
- 1.单例模式
- 1.1 饿汉模式的实现方法
- 1.2 懒汉模式的实现方法
- 2. 阻塞队列
- 2.1 引入生产消费者模型的意义：
- 2.2 阻塞队列put方法和take方法
- 2.3 实现阻塞队列--重点
- 3.定时器
- 3.1 定时器的使用
- 3.2 实现定时器
- 4 线程池
- 4.1 线程池的使用
- 4.2 实现一个简单的线程池 -- 重点
1.单例模式
单例模式：是一种设计模式，某个类，在一个进程中只创建出一个实例（对象），对代码进行一个更严格的校验和检查。
实现单例模式最基础的实现方式：
- 饿汉模式：
- 懒汉模式
1.1 饿汉模式的实现方法
单例模式中一种简单的写法，饿汉：形容创建实例非常迫切，实例是在类加载的时候就创建了，创建时机非常早，相当于程序一启动，实例就创建了。
class Singleton {
// 在这个Singleton 被加载的时候，就会初始化这个静态成员
private static Singleton instance = new Singleton();// instance 指向的这个对象，就是唯一的一个对象
public static Singleton getInstance() {
// 对于饿汉模式，getInstance直接返回Instance实例，这个操作本质上是读操作，在多线程情况下读取同一个变量是线程安全的
return instance;
}
private Singleton() {
}
}
public class ThreadDemo26 {
public static void main(String[] args) {
// Singleton singleton = new Singleton(); //
Singleton s = Singleton.getInstance();
Singleton s2 = Singleton.getInstance();
System.out.println(s == s2);
}
}
1.2 懒汉模式的实现方法
懒汉模式：创建实例的时机比较晚，只到第一次使用的时候才会创建实例。
注意：
- 在这个引用 指向唯一实例，这个引用先初始化为null，而不是立即创建实例
- 由于在 instance = new SingletonLazy(); 实例化的时候有读有写，在多线程下是不安全的，会出现指令重排序的线程安全问题，通过添加volatile解决。
- instance = new SingletonLazy(); 拆成三大步骤：
- 申请一段内存空间
- 在这内存上调用构造方法，创建出这个实例
- 把这个内存地址赋值给Instance引用变量
正常是123，但是在多线程下可能132，就会出现问题
- 如果InStance为null，就说明首次调用，首次调用就需要考虑到线程安全问题，如果非null，就说明是后续的调用，就不必加锁，即双重校验锁。
** volatile**：
- 保证内存可见性，每次访问变量都必须重新读取内存，而不会优化到寄存器/缓存中。
- 禁止指令重排序，针对这个volatile修饰的变量的读写操作相关指令，是不能被重排序的。
class SingletonLazy {
// 这个引用指向唯一实例，这个引用先初始化为null，而不是立即创建实例
// 在这里添加volatile 避免重排序引起的线程安全问题
private volatile static SingletonLazy instance = null;
private static Object locker = new Object();
// 在懒汉模式，有读也有写 instance = new SingletonLazy();,在多线程下是不安全的，且不是单例模式了，
// 1.通过加锁 synchronized 然后再 把if 和 new两个操作打包成一个原子的
public static SingletonLazy getInstance() {
// 2.如果Istance 为null，就说明首次调用，首次调用就需要考虑到线程安全问题
// 如果非null，就说明是后续的调用，就不必加锁
// 双重校验锁
if (instance == null) {
synchronized (locker) {
if (instance == null) {
instance = new SingletonLazy();// 3.还会出现指令重排序引起线程安全，通过添加volatile解决
/**instance = new SingletonLazy(); 拆成三大步骤
* 1.申请一段内存空间
* 2.在这个内存上调用构造方法，创建出这个实例
* 3.把这个内存地址赋值给Instance引用变量
* 正常是123，但是在多线程下可能132 就会出现问题
*/
}
}
}
return instance;
}
private SingletonLazy() {
}
}
public class ThreadDemo27 {
public static void main(String[] args) {
SingletonLazy s1 = SingletonLazy.getInstance();
SingletonLazy s2 = SingletonLazy.getInstance();
System.out.println(s1 == s2);
}
}
2. 阻塞队列
阻塞队列是一种特殊的队列，遵守先进先出的原则。
阻塞队列是一种线程安全的数据结构，包含两个特性：
- 当队列满的时候，继续入队列就会阻塞，直到有其他线程从队列中取走元素。
- 当队列空的时候，继续出队列也会阻塞，直到有其他线程往队列中插入元素。
基于阻塞队列，就可以实现 生产消费者模型（是一种多线程编程的方法）。
2.1 引入生产消费者模型的意义：
- 解耦合，即把代码的耦合程度，从高降低，在实际开发中，经常涉及到分布式系统，服务器整个功能不是由一个服务器全部完成的，而是每个服务器负责一部分功能，通过服务器之间的网络通信，最终完成整个功能。
如公网内 的电商网站客户端，获取到主页信息，机房内部网络中 有入口服务器A，用户服务器B，商品服务器C。在这个模型中，A代码就需要涉及到一些和B相关的操作，同样B也涉及A，A和C中的代码也相互涉及。如果B或者C挂了，对A的影响非常大，即为高耦合。
引入生产消费者模型之后就可降低耦合，即添加阻塞队列：
上述模型中，A和B、C都不是直接交互了，而是通过阻塞队列传话，如果B或者C挂了，对A的影响几乎没有。
- 削峰填谷：如下图模型，当请求多了，A的请求数量会增加很多，B用户服务器（找到对应用户信息）和C商品服务器（从数据库中匹配商品）都会有很大的影响。
添加阻塞队列之后：即使外界的请求出现峰值，队列没有业务逻辑，只是存储数据抗压能力很强。有效的防止了B和C被冲击挂掉。
2.2 阻塞队列put方法和take方法
- put(): put和offer都是入队列，而put带有阻塞功能，没带阻塞功能，队列满了会返回结果。
- take():取出元素的时候，带有阻塞功能，判定如果队列为空，就进行wait阻塞等待。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class ThreadDemo28 {
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);
// put和offer都是入队列 ，而 put带有阻塞功能，没带阻塞，队列满了会返回结果
queue.put("aaa");
// take 也带有阻塞功能
String elem = queue.take();
System.out.println("elem = " + elem);
elem = queue.take();
System.out.println("elem = " + elem);
}
}
2.3 实现阻塞队列–重点
1）先实现普通队列，基于数组来实现（环形队列），区分队列空和队列满：
- 浪费一个格子，定义一个head队头和tail队尾，tail最多走到head的前一个位置
- 引入size变量
2）再加上线程安全
3）再加上阻塞功能
- 队列满了，添加wait进行阻塞，队列不满，即出队列成功后进行notify唤醒
- 队列空了，再出队列，同样也需要阻塞，同样是在另一个入队列成功后的线程中唤醒
class MyBlockingQueue {
private String[] elems = null;
private int head = 0;
private int tail = 0;
private int size = 0;
public MyBlockingQueue(int capacity) {
elems = new String[capacity];
}
private Object locker = new Object();
public void put(String elem) throws InterruptedException {
synchronized (locker) {
// 使用while：wait可能会被提前唤醒（当条件还没满足，就被唤醒了）
while (size >= elems.length) {
// 队列满了 实现阻塞
locker.wait(); //在Java标准库推荐使用wait搭配while循环，多一/N次确认操作
}
// 如果加锁不包含 while判断队列是否为满，在多线程下就会导致当入队列就会多入一个
// 新的元素要放到 tail指向的位置上
elems[tail] = elem;
tail++;
// 如果队尾大于数组大小，让tail重新指向0下标，形成闭环
if (tail >= elems.length) { // tail = tail % elems.length
tail = 0;
}
size++;
// 入队列成功后进行唤醒
locker.notify();
}
}
public String take() throws InterruptedException {
String elem = null;
synchronized (locker) {
while (size == 0) {
//队列空了
// 实现阻塞
locker.wait();
}
// 取出 head 位置的元素并返回
elem = elems[head];
head++;
if (head >= elems.length) {
head = 0;
}
size--;
// 队列不满，即出队列成功之后，加上唤醒
locker.notify();
}
return elem;
}
}
public class ThreadDemo29 {
public static void main(String[] args) throws InterruptedException {
MyBlockingQueue queue = new MyBlockingQueue(1000);
// 生产者消费者模型（核心是阻塞队列,使用synchronized和wait/notify达到线程安全&阻塞） 不仅仅是一个线程
// 也可能是一个独立的服务器程序，甚至是一组服务器程序
// 生产者
Thread t1 = new Thread(() -> {
int n = 1;
while (true) {
try {
queue.put(n+" ");
System.out.println("生产元素：" + n);
n++;
//Thread.sleep(500); //生产一个消费一个
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
});
// 消费者
Thread t2 = new Thread(() -> {
while (true) {
try {
String n = queue.take();
System.out.println("消费元素：" + n);
Thread.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
});
t1.start();
t2.start();
}
}
3.定时器
在Java标准库中提供了定时器的使用，Timer类，核心方法是schedule()，它有两个参数，第一个参数是即将执行的任务代码，第二个是指定多长时间之后执行（单位ms）。
3.1 定时器的使用
import java.util.Timer;
import java.util.TimerTask;
// 运行完之后，进程没有结束，因为timer 里内置了线程（前台线程） timer不知道是否还要添加任务进来，
// 可以使用timer.cancel()来主动结束
public class ThreadDemo30 {
public static void main(String[] args) throws InterruptedException {
// 定义一个timer添加多任务，每个任务同时会带有一个时间
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
//时间到了之后，要执行的代码
System.out.println("hello timer 3000");
}
},3000);
timer.schedule(new TimerTask() {
@Override
public void run() {
System.out.println("hello timer 2000");
}
},2000);
timer.schedule(new TimerTask() {
@Override
public void run() {
System.out.println("hello timer 1000");
}
},1000);
System.out.println("hello main");
Thread.sleep(3000);
timer.cancel();
}
}
3.2 实现定时器
- 一个带优先级的阻塞队列（按从小到大的顺序）
- 队列中的每个元素是一个Task对象
- Task中带有一个时间属性，队首元素就是即将执行的
- 同时有一个worker线程一直扫描队首元素，看队首元素是否需要执行
import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.Timer;
// 通过这个类，来描述一个任务
class MyTimerTask implements Comparable<MyTimerTask> {
//在什么时间点来执行这个任务
// 此处约定这个time是一个ms 级别的时间戳
private long time;
public long getTime() {
return time;
}
// 实际任务要执行的代码
private Runnable runnable;
// delay 期望是一个相对时间
public MyTimerTask(Runnable runnable, long delay) {
this.runnable = runnable;
// 计算真正要执行任务的绝对时间（）
this.time = System.currentTimeMillis() + delay;
}
public void run() {
runnable.run();
}
@Override
public int compareTo(MyTimerTask o) {
return (int) (this.time - o.time);
}
}
// 通过这个类，来表示一个定时器
class MyTimer {
// 负责扫描任务队列，执行任务的线程
private Thread t = null;
// 任务队列
private PriorityQueue<MyTimerTask> queue = new PriorityQueue<>();
private Object locker = new Object();
// 把任务放进队列
public void schedule(Runnable runnable,long delay) {
synchronized (locker) {
MyTimerTask task = new MyTimerTask(runnable,delay);
queue.offer(task);
// 添加新的元素之后，就可以唤醒扫描线程的wait了
locker.notify();
}
}
public void cancel() {
//
}
// 构造方法，创建扫描线程，让扫描线程来完成判定和执行
public MyTimer() {
t = new Thread(() -> {
// 扫描线程就需要循环反复的扫描队首元素，然后判定队首元素是不是时间到了
// 如果没到时间，啥也没有
// 如果时间到了，就执行这个任务从队列中删除
while (true) {
try {
// 1. 解决线程安全问题
synchronized (locker) { //这里的代码执行速度很快，解锁之后立即又重新尝试加锁，导致
// 其他线程通过schedule想加锁，但是加不上 （即线程饿死） -》 引入wait/notify
if (queue.isEmpty()) {
// 暂时先不处理
locker.wait();
}
MyTimerTask task = queue.peek();
// 获取当前时间
long curTime = System.currentTimeMillis();
if (curTime >= task.getTime()) {
//当前时间已经达到了任务时间，就可以执行任务了
queue.poll();
task.run();
}else {
// 当前时间还没到任务时间，暂时不执行
// 不能使用sleep。会错过新的任务，也无法释放锁
//Thread.sleep(task.getTime() - curTime);
locker.wait(task.getTime() - curTime);
}
} // 释放锁的
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
// 启动线程
t.start();
}
}
public class ThreadDemo31 {
public static void main(String[] args) {
MyTimer timer = new MyTimer();
timer.schedule(new Runnable() {
@Override
public void run() {
System.out.println("hello 3000");
}
},3000);
timer.schedule(new Runnable() {
@Override
public void run() {
System.out.println("hello 2000");
}
},2000);
timer.schedule(new Runnable() {
@Override
public void run() {
System.out.println("hello 1000");
}
},1000);
System.out.println("hello main");
}
}
4 线程池
线程池：提前把需要用的线程，在线程池里准备好，需要用的时候就从池子里取，用完之后还给池子。
由于频繁的创建销毁进程，成本太高，引入了轻量级 进程，即线程，如果创建销毁线程的频率也进一步提高，此时线程的创建销毁开销也越来越大。
所以有两种优化此处的线程的创建和销毁：
- 引入轻量级 线程，即纤程/协程：本质是程序员在用户太代码进行调度，而不是靠内核的调度器调度，节省了调度上的开销。
- 线程池：把要使用的线程提前创建好，用完了之后也不要直接释放，而是放进线程池里以备下次使用，从而节省了创建销毁线程的开销。
引入问题：为什么从线程池里取线程就比系统申请更高效？
- 从线程池里取线程是纯用户态代码（可控的）
- 通过系统申请创建线程，就是需要内核来完成（不太可控）
4.1 线程池的使用
在Java标准库中，把ThreadPoolExecutor类表线程池，给封装 成 Executors 工厂类，工厂类：创建出不同的线程池对象（在内部把ThreadPoolExecutor创建好了并且设置不同的参数）。
ThreadPoolExecutor 线程池的参数：
- int corePoolSize :核心线程数，int maximunmPoolSize：最大线程数
- long keepAliveTime：保持存活时间，TimeUnit unit:时间单位（s,min,ms,hour）
- ThreadFactory threadFactor：线程工厂，通过这个工厂类来创建线程对象（Thread）
- RejectExecutionHandler handler:拒绝策略，在线程池中，有一个阻塞队列，能够容纳的元素有上限，当任务队列已经满了，如果继续往队列添加任务，线程池会进行下面4种操作：
Executor创建线程池的方式：
- newFixedThreadPool: 创建固定线程数的线程池
- newCachedThreadPool: 创建线程数目动态增长的线程池.
- newSingleThreadExecutor: 创建只包含单个线程的线程池.
- newScheduledThreadPool: 设定 延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer.
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadDemo32 {
public static void main(String[] args) {
// 创建线程池的时候，设定线程池的线程数量
ExecutorService service = Executors.newFixedThreadPool(4);
service.submit(new Runnable() {
@Override
public void run() {
System.out.println("hello");
}
});
}
}
** 创建线程池的时候，很多时候需要设定线程池的线程数量？**
- 不同的程序，能够设定的线程的数目是不同的，要具体问题具体分析， 一个线程是CPU密集型的任务（在线程run里面进行计算），还是IO密集型任务（在线程run里使用scanner读取用户的输入）
- 如果一个进程中所有的线程都是CPU密集型的，每个线程所有工作都是CPU上执行，此时线程数目就不应该超过N（CPU的逻辑核心数）。
- 如果一个进程中，所有线程都是IO密集型，每个线程的大部分工作都是等待IO，此时线程数目与那元超过N
- 由于程序的复杂性，所以需要通过实验/测试，即设定不同的线程数目，分别进行性能测试，衡量每种线程数目下，总的时间开销，和系统资源占用的开销，找到这之间的合适值。
4.2 实现一个简单的线程池 – 重点
- 提供构造方法，指定创建多少个线程
- 在构造方法中，把这些线程都创建好
- 有一个阻塞队列，能够持有要执行的任务
- 提供submit方法，可以添加新的任务
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
class MyThreadPoolExecutor {
private List<Thread> threadList = new ArrayList<>();
// 保存任务的队列
private BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000);
// 通过n来指定创建多少个线程
public MyThreadPoolExecutor(int n) {
for (int i = 0; i < n; i++) {
Thread t = new Thread(() -> {
// 线程把任务队列中的任务不停的取出来，并且进行执行
while (true) {
try {
// 此处的take 带有阻塞功能
// 如果队列为空，此时 take就会阻塞
Runnable runnable = queue.take();
// 取出一个执行一个
runnable.run();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
t.start();
threadList.add(t);
}
}
public void submit(Runnable runnable) throws InterruptedException {
queue.put(runnable);
}
}
public class ThreadDemo33 {
public static void main(String[] args) throws InterruptedException {
MyThreadPoolExecutor executor = new MyThreadPoolExecutor(4);
for (int i = 0; i < 1000; i++) {
// n是一个实事final变量 ，每次循环都是一个新的n，就可以被捕获
int n = i;
executor.submit(new Runnable() {
@Override
public void run() { // 回调函数访问当前外部作用域的变量就是变量捕获
// i 一值在变，把i改成成员变量或者 int n = i
System.out.println("执行任务" + n + ",当前线程为："+ Thread.currentThread().getName());
}
});
}
}
}
![python flask报错OSError: [WinError 10038] 在一个非套接字上尝试了一个操作。](https://img-blog.csdnimg.cn/direct/5bd21553e7c4449d96a267be09ae46aa.png)
