Python多线程和多进程

- 线程和进程是什么
- 进程间通信方式
- 线程间通信方式
- 死锁
多线程 Threading
- 什么是多线程
- 基本方法函数
- - join（）
  - Queue
- 继承使用线程
- 同步
- GIL锁
- 锁
多进程/多核 Multiprocessing
- 创建进程
- 进程池 pool
- 共享内存
- 进程锁

线程和进程是什么

进程

是程序的一次执行过程，是程序在执行过程中分配和管理资源的基本单位
每一个进程都有一个自己的地址空间`
至少有 5 种基本状态，它们是：初始态，执行态，等待状态，就绪状态，终止状态。

线程
是CPU`调度和分派的基本单位
与同属一个进程的其他的线程共享进程所拥有的全部资源,可以访问隶属进程的资源。

线程与进程的区别联系：

线程是进程的一部分。一个线程只能属于一个进程，而一个进程可以有多个线程，但至少有一个线程：线程是进程的一部分。
根本区别：进程是操作系统资源分配的基本单位，而线程是任务调度和执行的基本单位
所处环境：在操作系统中能同时运行多个进程（程序）；而在同一个进程（程序）中有多个线程同时执行（通过CPU调度，在每个时间片中只有一个线程执行）
内存分配：系统为每个进程分配不同的内存空间；而对线程而言，除了CPU外，系统不会为线程分配内存（线程所使用的资源来自其所属进程的资源），线程组之间只能共享资源

进程间通信方式

https://www.cnblogs.com/zgq0/p/8780893.html

无名管道（pipe）

半双工（Half Duplex）
只能用于具有亲缘关系的进程之间的通信（也是父子进程或者兄弟进程之间）
它可以看成是一种特殊的文件，对于它的读写也可以使用普通的read、write 等函数。但是它不是普通的文件，并不属于其他任何文件系统，并且只存在于内存中

FIFO 命名管道

可以在无关的进程之间交换数据
FIFO有路径名与之相关联，它以一种特殊设备文件形式存在于文件系统中
FIFO的通信方式类似于在进程中使用文件来传输数据，只不过FIFO类型文件同时具有管道的特性。在数据读出时，FIFO管道中同时清除数据，并且“先进先出”

消息队列（Message queue）
是消息的链接表，存放在内核中。一个消息队列由一个标识符（即队列ID）来标识。

消息队列是面向记录的，其中的消息具有特定的格式以及特定的优先级。
消息队列独立于发送与接收进程。进程终止时，消息队列及其内容并不会被删除。
消息队列可以实现消息的随机查询,消息不一定要以先进先出的次序读取,也可以按消息的类型读取。

信号量（semaphore）
它是一个计数器。信号量用于实现进程间的互斥与同步，而不是用于存储进程间通信数据。
共享内存（Shared Memory）
指两个或多个进程共享一个给定的存储区。信号量+共享内存通常结合在一起使用，信号量用来同步对共享内存的访问。

总结：
1.管道：速度慢，容量有限，只有父子进程能通讯
2.FIFO：任何进程间都能通讯，但速度慢
3.消息队列：容量受到系统限制，且要注意第一次读的时候，要考虑上一次没有读完数据的问题
4.信号量：不能传递复杂消息，只能用来同步
5.共享内存区：能够很容易控制容量，速度快，但要保持同步，比如一个进程在写的时候，另一个进程要注意读写的问题，相当于线程中的线程安全，当然，共享内存区同样可以用作线程间通讯，不过没这个必要，线程间本来就已经共享了同一进程内的一块内存

线程间通信方式

https://blog.csdn.net/qq_42473704/article/details/81942347

线程的生命周期

NEW	新建状态，刚刚创建完成还没开启的状态
RUNNABLE	可运行状态，有资格执行，可能在执行中，有可能不是在执行中
BLOCKED	锁阻塞状态，要等待其他线程释放锁对象
WAITING	无限等待，一个线程等待另一个线程执行一个（唤醒）动作
TIMED_WAITING	计时等待，这一状态一直保持到超过规定的时间，或者收到唤醒动作
TERMINATED	死亡状态，任务执行完毕的状态

线程通信方法

wait()	等待，让出cpu进入等待状态（如果一个线程内调用了该方法，那么该线程就停止运行，等待其他线程唤醒，或者其他线程调用notifAll方法）
notify()	唤醒，随机唤醒一个正在等待的线程，让其进入可运行状态（解除了调用wait方法线程的等待状态，让其变成可运行状态）
notifyAll()	唤醒所以进入等待状态的线程，让其都进入可运行状态

死锁

死锁产生的4个必要条件：
1、互斥：一个资源同一时刻只允许一个线程进行访问。
2、占有未释放：一个线程占有资源，且没有释放资源。
3、不可抢占：一个已经占有资源的线程无法抢占到其他线程拥有的资源。
4、循环等待：两个或者两个以上的线程，本身拥有资源，不释放资源，并且同时尝试获得其他线程所持有的资源，这种资源的申请关系形成一个闭环的链条。

解决办法：

死锁检测与恢复的方案是引入看门狗计数器。当线程正常运行的时候会每隔一段时间重置计数器，在没有发生死锁的情况下，一切都正常进行。一旦发生死锁，由于无法重置计数器导致定时器超时，这时程序会通过重启自身恢复到正常状态。
在进程获取锁的时候会严格按照对象id升序排列获取。避免死锁的主要思想是，单纯地按照对象id递增的顺序加锁不会产生循环依赖，而循环依赖是死锁的一个必要条件，从而避免程序进入死锁状态。

多线程 Threading

什么是多线程

单线程：
在程序当中，只有一个箭头指向代码从头到尾执行。

import time
def sorry():print("Sorry")time.sleep(1)if __name__ == '__main__':for i in range(5):sorry()

依次打印五个sorry，每个间隔1秒

多线程：
在一个脚本当中，一次同时运行多个程序。

import time
import threading
def sorry():print("Sorry")time.sleep(1)if __name__ == '__main__':for i in range(5):# Thread的参数为键 - 值对，target = 调用他的函数t = threading.Thread(target = sorry)   # 要用start启动线程t.start()

这是五个线程同时启动，五个“Sorry”同时打印出来。-- > 并发

基本方法函数

添加线程：

t = threading.Thread(target = thread_job)
t.start()

有多少个激活了的线程：

threading.active_count()

线程详情：

threading.enumerate()
threading.current_thread()

join（）

使得join以后的语句，必须要等到t线程的所有语句执行完以后才执行

import threading
import time
def thread_job():print("T1 start")for i in range(10):time.sleep(1)print("T1 finished")def main():t = threading.Thread(target = thread_job, name = 'T1')t.start()t.join()print("all done")
if __name__ == '__main__':main()

T1会总计sleep才会输出“ T1 finished”，应该会先print “all done”出来。
但是由于t.join（），所以后面的语句会等T1执行完才执行。

T1 start
T1 finished
all done

Queue

因为thread是没有返回值的，不能像函数一样返回。所以要用queue来储存结果，用于输出。

import threading
import time
from queue import Queuedef job(l, q):for i in range(len(l)):l[i] = l[i] ** 2q.put(l)   # 储存结果到queuedef multithreading(data):q = Queue()   # 储存结果threads = [] # 储存所有线程for i in range(4):t = threading.Thread(target = job, args = (data[i], q))    # 目标函数所接收的参数t.start()threads.append(t)for thread in threads:thread.join()res = []for _ in range(4):res.append(q.get()) # 得到结果print(res)if __name__ == '__main__':data = [[1, 2, 3], [3, 4, 5], [4, 4, 4], [5, 5, 5]]multithreading(data)

继承使用线程

要重写父类的run方法。

import time
import threading
class MyThread(threading.Thread):   # 括号里写要继承的父类# 重写父类的run方法def run(self):  for i in range(5):print(f"I am {self.name} {str(i)}")if __name__ == '__main__':t = MyThread()    # 实例化线程的类t.start()

五个线程同时打印：

I am Thread-1 0
I am Thread-1 1
I am Thread-1 2
I am Thread-1 3
I am Thread-1 4

每个线程一定会有一个名字，self.name --> Thread - 1
线程run（）方法结束，该线程结束
无法控制线程的调度，因为操作系统说了算。可以通过别的方式来影响线程的调度：互斥、死锁、条件变量。
线程的几种状态：新建 – 就绪 – 等待（阻塞）-- 运行态 – 死亡

同步

问题：
t1 t2两个线程，对一个全局变量num = 0进行增1运算，都对num修改十次
num永远为1，不会变成10.

两个不同的线程共享全局的资源，共享全局变量 --> 争夺！

使用多线程的时候，尽量避免全局变量，否则一定他要加锁lock（）

什么是同步：
协同步调，按照预定的先后顺序进行运行（如：单线程）。比如：你说完，我再说

锁实现同步

import time
import threading
num = 0 # 全局变量
# 创建一把锁
mutex = threading.Lock()    # 锁默认是开的
class MyThread(threading.Thread):def run(self):global num  # 声明要对全局变量进行修改mutexFlag = mutex.acquire()print(f"线程{self.name}的锁状态为{mutexFlag}")# 判断时候上锁成功if mutexFlag:num += 1time.sleep(1)print(f"{self.name} set num to {str(num)}")# 如果上锁以后不解开，就会“ 死锁 ”，程序卡在这里，不结束也不死亡# 解锁mutex.release()
def test():for i in range(5):t = MyThread()t.start()if __name__ == '__main__':test()

线程Thread-1的锁状态为True
Thread-1 set num to 1
线程Thread-2的锁状态为True
Thread-2 set num to 2
线程Thread-3的锁状态为True
Thread-3 set num to 3
线程Thread-4的锁状态为True
Thread-4 set num to 4
线程Thread-5的锁状态为True
Thread-5 set num to 5

锁：
好处：确保某段关键代码只能由一个线程从头到尾完整的执行
坏处：阻止了多线程并发运行，效率大大降低。由于可以存在多个锁，不同线程持有不同的锁，并且试图获取对方的锁，就可能会出现死锁。

GIL锁

GIL（Global Interpreter Lock）:

全局的互斥锁（mutex）
防止多线程同时操作一段字节码，为了Python解释器中原子操作的线程安全。

Python多线程不一定效率提升，因为python不是把任务平均分给每个线程同时做一个任务。实际上是靠GIL锁进行多线程，每次锁住一个线程进行任务，然后不断切换进程达成多线程的效果。python还是只能让一个线程在一个时间进行运算。

为了解决多线程之间数据完整性和状态同步的问题，设计为在任意时刻只有一个线程在解释器中运行。而当执行多线程程序时，由GIL来控制同一时刻只有一个线程能够运行。即Python中的多线程是表面多线程，也可以理解为fake多线程，不是真正的多线程。依靠解释器的分时复用，多个线程的代码，轮流被解释器执行。

普通解释：
并发：交替做不同事情的能力
并行：同时做不同事情的能力
专业术语：
并发：不同的代码块交替执行
并行：不同的代码块同时执行

当第一个线程处于输入输出阶段时，GIL release第一个线程，开始第二个线程，任务交给第二个线程。然后第二个线程用GIL锁住，不让其他线程进行运算。依次循环往复，仅仅是节省了读写的时间（进入读写 --> 切换进程）。

多个线程处理不同的任务效率提升明显（比如：一个线程负责聊天文字读取，一个线程负责文字输出）。
多个线程处理一个相同的任务是没什么提升。如果要提升，需要multiprocessing，多核运不同核之间不会受到GIL的影响

Python的多线程在多核CPU上，只对于IO密集型计算产生正面效果；而当有至少有一个CPU密集型线程存在，那么多线程效率会由于GIL而大幅下降。

避免受到GIL锁的影响
使用multiprocessing，多进程而不是多线程。每个进程有自己的独立的GIL，因此也不会出现进程之间的GIL争抢。
但是它的引入会增加程序实现时线程间数据通讯和同步的困难，multiprocess由于进程之间无法看到对方的数据。

锁

锁住第一个线程，等其执行完，再开始第二个线程。一般当要对多个线程的共同内存处理时，使用锁，完成对共同数据的多步加工。

import threading
def job1():global A, locklock.acquire()for i in range(10):A += 1print('job1', A)lock.release()
def job2():global A, locklock.acquire()for i in range(10):A += 10print('job2', A)lock.release()if __name__ == '__main__':lock = threading.Lock()   # 调用LockA = 0   # 全局变量，两个线程的公用内存t1 = threading.Thread(target=job1)t2 = threading.Thread(target=job2)t1.start()t2.start()t1.join()t2.join()

lock.acquire() 到 lock.release() 之间线程执行的代码时，不会允许其他线程执行。所以之间的代码执行完，才会轮到其他线程执行。

多进程/多核 Multiprocessing

利用多核处理器，真正完成同时运行不同的程序。

创建进程

进程的创建与线程完全一样。需要在 if __name__ == '__main__': 中创建。

import multiprocessing as mp
import threading as tddef job(a, b):print(a+b)if __name__ == '__main__':t1 = td.Thread(target = job, args = (1, 2))p1 = mp.Process(target = job, args = (1, 2))t1.start()p1.start()t1.join()p1.join()

同样需要使用 queue 来储存输出

import multiprocessing as mpdef job(q):res = 0passq.put(res)q = mp.Queue()
p1 = mp.Process(target = job, args = (q,)) # 如果只有一个参数，要留一个‘ , ’
res1 = q.get()

多线程、多进程对比

import multiprocessing as mp
import threading as td
import timedef job(q):res = 0for i in range(1000000):res += i+i**2+i**3q.put(res) # queuedef multicore():q = mp.Queue()p1 = mp.Process(target=job, args=(q,))p2 = mp.Process(target=job, args=(q,))p1.start()p2.start()p1.join()p2.join()res1 = q.get()res2 = q.get()print('multicore:' , res1+res2)def normal():res = 0for _ in range(2):for i in range(1000000):res += i+i**2+i**3print('normal:', res)def multithread():q = mp.Queue()t1 = td.Thread(target=job, args=(q,))t2 = td.Thread(target=job, args=(q,))t1.start()t2.start()t1.join()t2.join()res1 = q.get()res2 = q.get()print('multithread:', res1+res2)if __name__ == '__main__':st = time.time()normal()st1= time.time()print('normal time:', st1 - st)print("")multithread()st2 = time.time()print('multithread time:', st2 - st1)print("")multicore()print('multicore time:', time.time()-st2)

进程池 pool

用pool来完成输入，并储存输出。来替代queue

import multiprocessing as mpdef job(x):return x*xdef multicore():pool = mp.Pool(processes = 3) # processes代表使用几个核，默认使用所有核# map:一次把很多个值分配个很多个进程来运算res = pool.map(job, range(10))    # map（方法，给方法传入的参数）print(res)# >>> [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]# apply_async: 一次把一个值放入一个核内运算res = pool.apply_async(job, (2, ))    # 只能输入一个值print(res.get())# >>> 4# 使用迭代器可以完成多个输入multi_res = [pool.apply_async(job, (i, )) for i in range(10)]print([res.get() for res in multi_res]# >>> [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]if __name__ == '__main__':multicore()

共享内存

中间的L2 Cache就是上图四核CPU的共享内存。

import multiprocessing as mp# 这个value可以被所有进程所共享
value = mp.Value('d', 1)# 这个一维array可以被所有进程共享
array = mp.Array('i', [1, 2, 3])    # 只能是一维的array

通过这些共享内存，计算机的不同核可以相互交流。

进程锁

为了避免进程在共享内存中抢夺，用法同线程的锁。

import multiprocessing as mpdef job(v, num, l):l.acquire()for _ in range(10):time.sleep(0.1)v.value += numprint(v.value)l.release()def multicore():l = mp.Lock()v = mp.Value('i', 0)p1 = mp.Process(target = job, args = (v, 1, l))p2 = mp.Process(target = job, args = (v, 3, l))p1.start()p2.start()p1.join()p2.join()if __name__ == '__main__':multicore()

进程之间不互相干扰，p1执行完才执行p2