当前位置:   article > 正文

多线程(Threading)和多进程(Multiprocessing)_threading和multiprocessing

threading和multiprocessing


线程和进程是什么

进程

  • 是程序的一次执行过程,是程序在执行过程中分配和管理资源的基本单位
  • 每一个进程都有一个自己的地址空间`
  • 至少有 5 种基本状态,它们是:初始态,执行态,等待状态,就绪状态,终止状态。
    在这里插入图片描述
    线程
  • 是CPU`调度和分派的基本单位
  • 与同属一个进程的其他的线程共享进程所拥有的全部资源,可以访问隶属进程的资源。

线程与进程的区别联系:

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

对于单核的CPU来说,真的可以做到真正的多线程并发吗?
对于多核的CPU电脑来说,真正的多线程并发是没问题的。4核CPU表示同一个时间点上,可以真正的有4个进程并发执行。

单核的CPU表示只有一个大脑:
不能够做到真正的多线程并发,但是可以做到给人一种“多线程并发”的感觉。
对于单核的CPU来说,在某一个时间点上实际上只能处理一件事情,但是由于CPU的处理速度极快,多个线程之间频繁切换执行,给别人的感觉是:多个事情同时在做!!!
eg.
线程A:播放音乐;线程B:运行游戏
线程A和线程B频繁切换执行,人类会感觉音乐一直在播放,游戏一直在运行,给我们的感觉是同时并发的。(因为计算机的速度很快,我们人的眼睛很慢,所以才会感觉是多线程!)


进程间通信方式

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

  1. 无名管道(pipe)
  • 半双工(Half Duplex)
  • 只能用于具有亲缘关系的进程之间的通信(也是父子进程或者兄弟进程之间)
  • 它可以看成是一种特殊的文件,对于它的读写也可以使用普通的read、write 等函数。但是它不是普通的文件,并不属于其他任何文件系统,并且只存在于内存中
  1. FIFO 命名管道
  • 可以在无关的进程之间交换数据
  • FIFO有路径名与之相关联,它以一种特殊设备文件形式存在于文件系统中
  • FIFO的通信方式类似于在进程中使用文件来传输数据,只不过FIFO类型文件同时具有管道的特性。在数据读出时,FIFO管道中同时清除数据,并且“先进先出”
  1. 消息队列(Message queue)
    是消息的链接表,存放在内核中。一个消息队列由一个标识符(即队列ID)来标识。
  • 消息队列是面向记录的,其中的消息具有特定的格式以及特定的优先级。
  • 消息队列独立于发送与接收进程。进程终止时,消息队列及其内容并不会被删除。
  • 消息队列可以实现消息的随机查询,消息不一定要以先进先出的次序读取,也可以按消息的类型读取。
  1. 信号量(semaphore)
    它是一个计数器。信号量用于实现进程间的互斥与同步,而不是用于存储进程间通信数据。
  2. 共享内存(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、循环等待:两个或者两个以上的线程,本身拥有资源,不释放资源,并且同时尝试获得其他线程所持有的资源,这种资源的申请关系形成一个闭环的链条。

解决办法:

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

Python多线程 Threading

什么是多线程

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

import time
def sorry():
    print("Sorry")
    time.sleep(1)

if __name__ == '__main__':
    for i in range(5):
        sorry()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

依次打印五个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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

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

基本方法函数

添加线程:

t = threading.Thread(target = thread_job)
t.start()
  • 1
  • 2

有多少个激活了的线程:

threading.active_count()
  • 1

线程详情:

threading.enumerate()
threading.current_thread()
  • 1
  • 2

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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

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

T1 start
T1 finished
all done
  • 1
  • 2
  • 3

Queue

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

import threading
import time
from queue import Queue

def job(l, q):
    for i in range(len(l)):
        l[i] = l[i] ** 2
    q.put(l)	# 储存结果到queue

def 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)
  • 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

继承使用线程

要重写父类的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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

五个线程同时打印:

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
  • 1
  • 2
  • 3
  • 4
  • 5
  1. 每个线程一定会有一个名字,self.name --> Thread - 1
  2. 线程run()方法结束,该线程结束
  3. 无法控制线程的调度,因为操作系统说了算。可以通过别的方式来影响线程的调度:互斥、死锁、条件变量。
  4. 线程的几种状态:新建 – 就绪 – 等待(阻塞)-- 运行态 – 死亡

同步

问题:
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 += 1
            time.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()
  • 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
线程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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

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


GIL锁

GIL(Global Interpreter Lock):

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, lock
    lock.acquire()
    for i in range(10):
        A += 1
        print('job1', A)
    lock.release()
def job2():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 10
        print('job2', A)
    lock.release()

if __name__ == '__main__':
    lock = threading.Lock()	# 调用Lock
    A = 0   # 全局变量,两个线程的公用内存
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.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

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

Python多进程/多核 Multiprocessing

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

创建进程

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

import multiprocessing as mp
import threading as td

def 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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

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

import multiprocessing as mp

def job(q):
	res = 0
	pass
	q.put(res)
	
q = mp.Queue()
p1 = mp.Process(target = job, args = (q,)) # 如果只有一个参数,要留一个‘ , ’
res1 = q.get()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

多线程、多进程对比

import multiprocessing as mp
import threading as td
import time

def job(q):
    res = 0
    for i in range(1000000):
        res += i+i**2+i**3
    q.put(res) # queue

def 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 = 0
    for _ in range(2):
        for i in range(1000000):
            res += i+i**2+i**3
    print('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)
  • 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
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53

进程池 pool

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

import multiprocessing as mp

def job(x):
    return x*x

def 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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24

共享内存

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

在这里插入图片描述

import multiprocessing as mp

# 这个value可以被所有进程所共享
value = mp.Value('d', 1)

# 这个一维array可以被所有进程共享
array = mp.Array('i', [1, 2, 3])    # 只能是一维的array
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

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

进程锁

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

import multiprocessing as mp

def job(v, num, l):
    l.acquire()
    for _ in range(10):
        time.sleep(0.1)
        v.value += num
        print(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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

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

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/Monodyee/article/detail/74881
推荐阅读
相关标签
  

闽ICP备14008679号