linux 多线程的实现的基本原理
时间:2019-05-28 14:17 来源:linux.it.net.cn 作者:IT
1. linux 多线程的基本概念
linux 是多用户、多任务的并发执行；所谓的并发是通过多进程、多线程来实现的；
1). 其中多进程有3种方式：
单机多实例(机器复用，一台机器启动多个进程，每个进程干自己的事情)
多进程(比如24core --> 启动24core) ：nginx，通过(主进程(master)-->从进程(slave)的方式来调度和完成任务的分发，从而实现nginx 并发处理多个需求)
进程池：fork 多个，组装成一个进程池，然后通过主进程进行并发task 的调度
2). 其中多线程有2种方式:
多线程:启动一个进程(线程必须依附于进程来实现，通过启动进程来启动线程)；也就是说一个进程至少有一个线程(每个人理解不用，这个句话也不一定正确，可以认为线程是最基本的执行单元，进程是最基本的程序单元【进程可以管理task、cpu、存储等】)；一个进程可以启动多个线程来实现并发处理多个task；
线程池：启动、准备一个线程池(线程池的实现原理 + task 调度原理是？)
2. 多线程的代码实现(java 语言来学习)
1). 代码实现：
#实现一个线程类(一般用runable 来实现)
public class MyThread extends Thread{
private String name;
public MyThread(string name){
this.name = name;
}
#此线程的执行体都写在run方法里面,线程干的事情(可以在run里面写一个httpserver 服务器，并发处理一些http的请求)
public void run()
{
for(int i=0;i<100;i++)
{
Systen.out.printIn(name+":" +i);
}
super.run();
}
2). 启动线程
# 新建一个线程启动(干一件事情)
public class ThreadDemo01
{
public static void man(String[] args)
{
MyThread t1 = new MyThread("A");
MyThread t2 = new MyThread("B");
MyThread t3 = new MyThread("C");
# 启动线程(不能用run方法，要用start方法)
t1.start()
t2.start()
t3.start()
}
}
3). 执行多线程的处理结果
# 分析结果
a). 这里启动的三个线程干同样的事情，也可以干不同的事情，只需要实现三个线程类，把要干的逻辑写到各自的run方法即可
b). 这里启动了三个线程干同样的事情类似于实现了一个最简单的线程池，当然没有考虑各种容错、用户体验、api 逻辑等等，但线程池就应该是这么个意思
c). 三个线程会并发处理task，cpu 会在三个线程中进行切换[至于三个线程之间怎么切换：现在还不清楚，但是代码可以做一些控制：比如sleep 礼让 也可以设置权重吧]
3. 线程的常用方法和概念
1). 线程锁
2). 线程的状态/生命周期
3). 线程的优先级
4). 线程的同步
5). 线程的常用方法
4. 线程池的代码实现(java 代码学习)
上面的线程的使用是在进程中需要并发处理一块task时，我们新建一个线程，这个线程处理的事件(run(方法写的逻辑))如果是一个“短时间" 内可以完成的task，那task 完成之后们需要关闭线程，下次再来任务时(重新执行我们的程序代码，也就是我们的程序接受到一个相同的或者不同的task)我们再次新建线程；这样很方便但存在一个问题：在并发成处理很多的请求，并发率非常高的情况下，我们每处理一个请求要新建一个线程、处理完后shutdown 一个线程;这样频繁的新建、关闭线程会占用大量的资源；这时我们采用线程池来解决这个问题，就是建立线程后，此线程处理完请求(完成task) 后不shutdown ，而是让后面的task(请求)继续复用这个链接；当链接空闲时间超过【一个设置的数值】后会shutdown 退出。
1). java 线程类: ThreadPoolExecutor 类
这个类是线程池的核心类，创建线程池就需要实现这个类，这个类有4个构造方法，每个构造方法都可以实现这个类；这个类也是层层继承一直到最基本的接口类:
ThreadPoolExecuto--继承-->AbstractExecutorService --继承-->ExecutorService --继承-->Executor;
# 这个类的构造方法
public class ThreadPoolExecutor extends AbstractExecutorService {
.....
#一共4个构造法方法，每个方法参数不同，根据自己需要进行选择；前三个都是调用的第4个基础方法
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
...
}
2). 构造方法里面参数的意思:
corePoolSize: 核心线程池的大小,当线程池的线程数量达到这个数值后会进行增加线程池的个数
maximumPoolSize: 线程池最大的线程数量，当超过corePoolsize，可以新增加线程的个数；但总数不能超过这个数量上限
KeepAliveTime: 线程在闲置状态保留的最大时间，一旦超过这个时间会关闭这个线程(粒度: 天、小时、秒、毫秒)
workQueue: 一个阻塞队列，用来存储等待执行的task；这个队列有三个方式:
#workQueue 的三种工作方式
ArrayBlockingQueue; #数组的方式来存储队列
LinkedBlockingQueue; #链表的方式来存储队列
SynchronousQueue; #同步队列,这个队列没有容量的概念，必须是take和put 同时进行；take一个task，必须put一个task(不是很理解:如果后面没有task了怎么办？)
# blockingqueue 的含义
block 阻塞，就是阻塞队列，take()和put() 都需要阻塞，无论读合写都只能一个线程来搞
#put() /take() LinkBlockingQueue 是 双锁；加锁的目的就是不要出现脏数据，比如一个task已经被一个线程take(),但另外一个线程也take()会出现问题;
# 加锁就要想避免出现死锁的方案
RejectedExecutionHandler: 拒绝策略：当缓存队列已满，线程池中的线程数量达到maxinumPollSize 时如果再有task 进来，这时就要拒绝任务啦：
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
3). 提交任务给线程池
线程池创建好以后，我们需要把task 提交给线程池来实现；提交有2种方案:
1). 有一个专门负责调度的线程(一般称之为主线程)，会监听线程池线程空闲状态、workQueue 状态；如果线程池有空闲的状态，主线程会从workQueue 里面拿取一个task,然后submit 给ThreadPool；如果ThreadPool 里面没有空闲的线程，主线程收到task 会把这个task 写入到workQueue，如果这个WorkQueue 也满了，而且ThreadPool 的线程数量也达到了maxinumSize，则就会拒绝请求(根据策略来决定)
2). ThreadPool 里面的线程空闲之后自己去workQueue 里面获取task，执行完task后循环即可；但是还是需要一个专门的线程来接收task，并把task 放入workQueue ,一旦满足拒绝的条件跟进拒绝策略进行拒绝请求。
5. 线程池的实现原理
5.1 线程池的状态
RUNNING: 创建线程池，初始化之后就是RUNNING
SHUTDOWN 状态: 收到shutdown 命令
STOP 状态: 收到shutdownnow 命令
5.2 任务执行(task)
先了解下ThreadPoolExecutor 类的重要的成员变量
private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock(); //线程池的主要状态锁，对线程池状态（比如线程池大小
//、runState等）的改变都要使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集
private volatile long keepAliveTime; //线程存货时间
private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间
private volatile int corePoolSize; //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int maximumPoolSize; //线程池最大能容忍的线程数
private volatile int poolSize; //线程池中当前的线程数
private volatile RejectedExecutionHandler handler; //任务拒绝策略
private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程
private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数
private long completedTaskCount; //用来记录已经执行完毕的任务个数
最重要的是任务提交的方法: execute/submit
submit 提交方法的实质也是调用的execute()方法，可以研究下execute()的原理
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
5.3 线程池的初始化
默认情况下线程池创建后里面是没有线程的，当第一个task 来时进行创建线程；
如果想在创建线程池后就有线程，需要调用预先创建线程的方法
5.4 线程池的关闭
ThreadPoolExecutor 提供了2个方法来关闭连接池，
shutdown： 不再接收新任务，但要等workQueue 里面的task 都执行完毕后才可以关闭线程池
shutdownnow: 丢弃掉queue 里面的task，直接关闭
5.5 线程池容量的动态调整
threadCount == corePoolSize && workqueue == full,此时就会创建新线程进行工作；
threadCout == maxinumPoolSize && workqueue ==full ,直接丢弃新的请求
6. 线程池的示例
public class Test {
#创建一个main 方法
public static void main(String[] args) {
#创建一个线程池的对象
ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<Runnable>(5));
for(int i=0;i<15;i++){
#实现线程的一个对象(因为线程类是同一个，所以for 循环里面所有的对象都是干同样的事情)
MyTask myTask = new MyTask(i);
# 把对象(线程里面的run 方法的task )提交给线程池(调用 execute())
executor.execute(myTask);
#打印相关信息
System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
}
executor.shutdown();
}
}
#实现一个线程类
class MyTask implements Runnable {
private int taskNum;
public MyTask(int num) {
this.taskNum = num;
}
#线程中的run 方法[1). 打印执行task 2). sleep 4s 观察线程池的并发 3). 打印执行完毕 ]
@Override
public void run() {
System.out.println("正在执行task "+taskNum);
try {
Thread.currentThread().sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("task "+taskNum+"执行完毕");
}
}
(责任编辑：IT)
1. linux 多线程的基本概念
linux 是多用户、多任务的并发执行；所谓的并发是通过多进程、多线程来实现的；
1). 其中多进程有3种方式：
单机多实例(机器复用，一台机器启动多个进程，每个进程干自己的事情)
多进程(比如24core --> 启动24core) ：nginx，通过(主进程(master)-->从进程(slave)的方式来调度和完成任务的分发，从而实现nginx 并发处理多个需求)
进程池：fork 多个，组装成一个进程池，然后通过主进程进行并发task 的调度
2). 其中多线程有2种方式:
多线程:启动一个进程(线程必须依附于进程来实现，通过启动进程来启动线程)；也就是说一个进程至少有一个线程(每个人理解不用，这个句话也不一定正确，可以认为线程是最基本的执行单元，进程是最基本的程序单元【进程可以管理task、cpu、存储等】)；一个进程可以启动多个线程来实现并发处理多个task；
线程池：启动、准备一个线程池(线程池的实现原理 + task 调度原理是？)
2. 多线程的代码实现(java 语言来学习)
1). 代码实现：
#实现一个线程类(一般用runable 来实现)
public class MyThread extends Thread{
private String name;
public MyThread(string name){
this.name = name;
}
#此线程的执行体都写在run方法里面,线程干的事情(可以在run里面写一个httpserver 服务器，并发处理一些http的请求)
public void run()
{
for(int i=0;i<100;i++)
{
Systen.out.printIn(name+":" +i);
}
super.run();
}
2). 启动线程
# 新建一个线程启动(干一件事情)
public class ThreadDemo01
{
public static void man(String[] args)
{
MyThread t1 = new MyThread("A");
MyThread t2 = new MyThread("B");
MyThread t3 = new MyThread("C");
# 启动线程(不能用run方法，要用start方法)
t1.start()
t2.start()
t3.start()
}
}
3). 执行多线程的处理结果
# 分析结果
a). 这里启动的三个线程干同样的事情，也可以干不同的事情，只需要实现三个线程类，把要干的逻辑写到各自的run方法即可
b). 这里启动了三个线程干同样的事情类似于实现了一个最简单的线程池，当然没有考虑各种容错、用户体验、api 逻辑等等，但线程池就应该是这么个意思
c). 三个线程会并发处理task，cpu 会在三个线程中进行切换[至于三个线程之间怎么切换：现在还不清楚，但是代码可以做一些控制：比如sleep 礼让 也可以设置权重吧]
3. 线程的常用方法和概念
1). 线程锁
2). 线程的状态/生命周期
3). 线程的优先级
4). 线程的同步
5). 线程的常用方法
4. 线程池的代码实现(java 代码学习)
上面的线程的使用是在进程中需要并发处理一块task时，我们新建一个线程，这个线程处理的事件(run(方法写的逻辑))如果是一个“短时间" 内可以完成的task，那task 完成之后们需要关闭线程，下次再来任务时(重新执行我们的程序代码，也就是我们的程序接受到一个相同的或者不同的task)我们再次新建线程；这样很方便但存在一个问题：在并发成处理很多的请求，并发率非常高的情况下，我们每处理一个请求要新建一个线程、处理完后shutdown 一个线程;这样频繁的新建、关闭线程会占用大量的资源；这时我们采用线程池来解决这个问题，就是建立线程后，此线程处理完请求(完成task) 后不shutdown ，而是让后面的task(请求)继续复用这个链接；当链接空闲时间超过【一个设置的数值】后会shutdown 退出。
1). java 线程类: ThreadPoolExecutor 类
这个类是线程池的核心类，创建线程池就需要实现这个类，这个类有4个构造方法，每个构造方法都可以实现这个类；这个类也是层层继承一直到最基本的接口类:
ThreadPoolExecuto--继承-->AbstractExecutorService --继承-->ExecutorService --继承-->Executor;
# 这个类的构造方法
public class ThreadPoolExecutor extends AbstractExecutorService {
.....
#一共4个构造法方法，每个方法参数不同，根据自己需要进行选择；前三个都是调用的第4个基础方法
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
...
}
2). 构造方法里面参数的意思:
corePoolSize: 核心线程池的大小,当线程池的线程数量达到这个数值后会进行增加线程池的个数
maximumPoolSize: 线程池最大的线程数量，当超过corePoolsize，可以新增加线程的个数；但总数不能超过这个数量上限
KeepAliveTime: 线程在闲置状态保留的最大时间，一旦超过这个时间会关闭这个线程(粒度: 天、小时、秒、毫秒)
workQueue: 一个阻塞队列，用来存储等待执行的task；这个队列有三个方式:
#workQueue 的三种工作方式
ArrayBlockingQueue; #数组的方式来存储队列
LinkedBlockingQueue; #链表的方式来存储队列
SynchronousQueue; #同步队列,这个队列没有容量的概念，必须是take和put 同时进行；take一个task，必须put一个task(不是很理解:如果后面没有task了怎么办？)
# blockingqueue 的含义
block 阻塞，就是阻塞队列，take()和put() 都需要阻塞，无论读合写都只能一个线程来搞
#put() /take() LinkBlockingQueue 是 双锁；加锁的目的就是不要出现脏数据，比如一个task已经被一个线程take(),但另外一个线程也take()会出现问题;
# 加锁就要想避免出现死锁的方案
RejectedExecutionHandler: 拒绝策略：当缓存队列已满，线程池中的线程数量达到maxinumPollSize 时如果再有task 进来，这时就要拒绝任务啦：
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
3). 提交任务给线程池
线程池创建好以后，我们需要把task 提交给线程池来实现；提交有2种方案:
1). 有一个专门负责调度的线程(一般称之为主线程)，会监听线程池线程空闲状态、workQueue 状态；如果线程池有空闲的状态，主线程会从workQueue 里面拿取一个task,然后submit 给ThreadPool；如果ThreadPool 里面没有空闲的线程，主线程收到task 会把这个task 写入到workQueue，如果这个WorkQueue 也满了，而且ThreadPool 的线程数量也达到了maxinumSize，则就会拒绝请求(根据策略来决定)
2). ThreadPool 里面的线程空闲之后自己去workQueue 里面获取task，执行完task后循环即可；但是还是需要一个专门的线程来接收task，并把task 放入workQueue ,一旦满足拒绝的条件跟进拒绝策略进行拒绝请求。
5. 线程池的实现原理
5.1 线程池的状态
RUNNING: 创建线程池，初始化之后就是RUNNING
SHUTDOWN 状态: 收到shutdown 命令
STOP 状态: 收到shutdownnow 命令
5.2 任务执行(task)
先了解下ThreadPoolExecutor 类的重要的成员变量
private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock(); //线程池的主要状态锁，对线程池状态（比如线程池大小
//、runState等）的改变都要使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集
private volatile long keepAliveTime; //线程存货时间
private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间
private volatile int corePoolSize; //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int maximumPoolSize; //线程池最大能容忍的线程数
private volatile int poolSize; //线程池中当前的线程数
private volatile RejectedExecutionHandler handler; //任务拒绝策略
private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程
private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数
private long completedTaskCount; //用来记录已经执行完毕的任务个数
最重要的是任务提交的方法: execute/submit
submit 提交方法的实质也是调用的execute()方法，可以研究下execute()的原理
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
5.3 线程池的初始化
默认情况下线程池创建后里面是没有线程的，当第一个task 来时进行创建线程；
如果想在创建线程池后就有线程，需要调用预先创建线程的方法
5.4 线程池的关闭
ThreadPoolExecutor 提供了2个方法来关闭连接池，
shutdown： 不再接收新任务，但要等workQueue 里面的task 都执行完毕后才可以关闭线程池
shutdownnow: 丢弃掉queue 里面的task，直接关闭
5.5 线程池容量的动态调整
threadCount == corePoolSize && workqueue == full,此时就会创建新线程进行工作；
threadCout == maxinumPoolSize && workqueue ==full ,直接丢弃新的请求
6. 线程池的示例
public class Test {
#创建一个main 方法
public static void main(String[] args) {
#创建一个线程池的对象
ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<Runnable>(5));
for(int i=0;i<15;i++){
#实现线程的一个对象(因为线程类是同一个，所以for 循环里面所有的对象都是干同样的事情)
MyTask myTask = new MyTask(i);
# 把对象(线程里面的run 方法的task )提交给线程池(调用 execute())
executor.execute(myTask);
#打印相关信息
System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
}
executor.shutdown();
}
}
#实现一个线程类
class MyTask implements Runnable {
private int taskNum;
public MyTask(int num) {
this.taskNum = num;
}
#线程中的run 方法[1). 打印执行task 2). sleep 4s 观察线程池的并发 3). 打印执行完毕 ]
@Override
public void run() {
System.out.println("正在执行task "+taskNum);
try {
Thread.currentThread().sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("task "+taskNum+"执行完毕");
}
}
(责任编辑：IT) |