python中多进程和多线程的区别

Posted 2023-05-08

什么是线程、进程?
进程(process)与线程(thread)是操作系统的基本概念，它们比较抽象，不容易掌握。
关于这两者，最经典的一句话就是“进程是资源分配的最小单位，线程是CPU调度的最小单位”，线程是程序中一个单一的顺序控制流程，进程内一个相对独立的、可调度的执行单元，是系统独立调度和分配CPU的基本单位指运行中的程序的调度单位，在单个程序中同时运行多个线程完成不同的工作，称为多线程。
进程与线程的区别是什么?
进程是资源分配的基本单位，所有与该进程有关的资源，都被记录在进程控制块PCB中，以表示该进程拥有这些资源或正在使用它们，另外，进程也是抢占处理机的调度单位，它拥有一个完整的虚拟地址空间，当进程发生调度时，不同的进程拥有不同的虚拟地址空间，而同一进程内的不同线程共享同一地址空间。
与进程相对应的，线程与资源分配无关，它属于某一个进程，并与进程内的其他线程一起共享进程的资源，线程只由相关堆栈(系统栈或用户栈)寄存器和线程控制表TCB组成，寄存器可被用来存储线程内的局部变量，但不能存储其他线程的相关变量。
通常在一个进程中可以包含若干个线程，它们可以利用进程所拥有的资源，在引入线程的操作系统中，通常都是把进程作为分配资源的基本单位，而把线程作为独立运行和独立调度的基本单位。
由于线程比进程更小，基本上不拥有系统资源，所以对它的调度所付出的开销就会小得多，能更高效的提高系统内多个程序间并发执行的程度，从而显著提高系统资源的利用率和吞吐量。
因而近年来推出的通用操作系统都引入了线程，以便进一步提高系统的并发性，并把它视为现代操作系统的一个重要指标。 参考技术A 多线程可以共享全局变量，多进程不能。
多线程中，所有子线程的进程号相同；多进程中，不同的子进程进程号不同。
多进程比线程更易用，但是消耗更多的内存。本回答被提问者采纳

python中多线程

多线程

• 什么是多线程
• 开启线程的两种方式
• 进程和线程的区别
• Thread对象的其他属性和方法
• 守护线程
• 死锁现象与递归锁
• 信号量、Event定时器
• 线程Queue
• 进程池和线程池

什么是多线程

在传统意义上，每个进程有一个地址空间，而且默认就会有一个控制线程。

线程顾名思义，就是一条流水线工作的过程（流水线的工作需要电源，电源就相当于CPU），而一条流水线必须属于一个车间，一个车间的工作过程是一个进程，车间负责把资源整合到一起，是一个资源单位，而一个车间至少要有一条流水线。

所以，进程只是用来把资源整合到一起，而线程才是CPU上的执行单位。

多线程（多个控制线程）的概念是：在一个进程中存在多个线程，多个线程共享该进程的地址空间，相当于一个车间内有多条流水线，都公用一个车间的资源。

so：进程之间数据不共享，线程之间数据是共享的。

开启线程的两种方式

# 第一种
from threading import Thread
import time

def task(name):
    print(‘%s is running.‘%name)
    time.sleep(2)
    print(‘%s is done.‘%name)

if __name__ == ‘__main__‘:
    t = Thread(target=task,args=(‘子线程‘,))
    t.start()
    print(‘主线程‘)
# 运行结果
子线程 is running.
主线程
子线程 is done.

---------------------------------------------------
# 第二种
from threading import Thread
import time

class MyThread(Thread):
    def __init__(self,name):
        super(MyThread, self).__init__()
        self.name = name
    def run(self):
        print(‘%s is running.‘%self.name)
        time.sleep(2)
        print(‘%s is done.‘%self.name)

if __name__ == ‘__main__‘:
    t = MyThread(‘子线程‘)
    t.start()
    print(‘主线程‘)
# 运行结果
子进程 is running.
主进程
子进程 is done.

进程和线程的区别

• 开进程的开销远大于开线程
• 同一进程内的多个线程共享该进程的地址空间
• PID

验证开进程的开销远大于开线程

# 开启多进程
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self,name):
        super(MyProcess, self).__init__()
        self.name = name
    def run(self):
        print(‘%s is running.‘%self.name)
        time.sleep(2)
        print(‘%s is done.‘%self.name)

if __name__ == ‘__main__‘:
    p = MyProcess(‘子进程‘)
    p.start()
    print(‘主进程‘)

# 开启多线程
from threading import Thread
import time

class MyThread(Thread):
    def __init__(self,name):
        super(MyThread, self).__init__()
        self.name = name
    def run(self):
        print(‘%s is running.‘%self.name)
        time.sleep(2)
        print(‘%s is done.‘%self.name)

if __name__ == ‘__main__‘:
    p = MyThread(‘子线程‘)
    p.start()
    print(‘主线程‘)

如果这两段代码你分别运行一遍，你就可以明显的看出来，在开启多进程的时候，首先打印的是主进程，因为子进程此时在申请内存地址，然后才会把子进程打印出来；

而在开启多线程的时候，会立马把这个子进程首先打印出来。

同一个进程内的多个进程共享该进程的地址空间

说白了就是线程之间数据共享，因为之前做过了进程之间数据不共享的实验，所以这次就写一个线程之间共享的代码：

from threading import Thread
from multiprocessing import Process
import time

n = 100

def task(name):
    global n
    n = 99
    print(‘[%s]内n的值为<%s>‘%(name,n))

if __name__ == ‘__main__‘:
    t = Thread(target=task,args=(‘子线程‘,))
    t.start()
    print(‘主线程n的值为%s‘%n)

# 运行结果为
[子线程]内n的值为<99>
主线程n的值为99

足以证明线程之间数据是共享的。

瞅一眼pid

# 查看主进程和子进程的pid
from multiprocessing import Process,current_process
import time

def task():
    print(‘子进程:‘,current_process().pid)
    time.sleep(2)

if __name__ == ‘__main__‘:
    p = Process(target=task,name=‘子进程‘)
    p.start()
    print(‘主进程:‘,current_process().pid)
    
# 运行结果为:
主进程: 7912
子进程: 12092

# 查看主线程和子线程的pid
from threading import Thread,current_thread
import time,os

def task():
    print(‘子线程:‘,os.getpid())
    time.sleep(2)

if __name__ == ‘__main__‘:
    t = Thread(target=task,name=‘Thread子线程‘)
    t.start()
    print(‘主线程:‘,os.getpid())
    
# 运行结果为：
子线程: 9060
主线程: 9060
    
因为在多线程中，每个线程都是平级的，没有子线程的概念，为了方便理解，所以叫为‘子线程‘

Thread对象的属性和方法

name和getName()方法，其中name设置线程的名字，getName获取线程的名字

from threading import Thread,current_thread
import time,os

def task():
    print(‘子线程名为:‘,current_thread().getName())  # 获取当前线程的名字
    time.sleep(2)

if __name__ == ‘__main__‘:
    t = Thread(target=task,name=‘Thread子线程‘)  # 设置子线程的名字
    t.start()
    print(‘主线程‘)
    
# 运行结果
主线程
子线程名为: Thread子线程

join()方法和is_alive()方法

join()方法让主线程等待子线程完成之后再进行，is_alive()方法检测当前线程是否存活

from threading import Thread,current_thread
import time,os

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

if __name__ == ‘__main__‘:
    t = Thread(target=task,args=(‘子进程1‘,))
    t.start()
    print(t.is_alive())  # 判断是否存活
    t.join()  #让主进程等待
    print(‘主线程‘)
    print(t.is_alive())  # 判断是否存活
    
# 运行结果为
子进程1 is running
True
子进程1 is done
主线程
False

active_count()检查存活的线程数量

from threading import Thread,current_thread,active_count  # 导入这个模块
import time,os

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

if __name__ == ‘__main__‘:
    t1 = Thread(target=task,args=(‘子进程1‘,))
    t2 = Thread(target=task,args=(‘子进程2‘,))
    t1.start()
    t2.start()  
    print(active_count())  # 判断当前线程数
    t1.join()
    t2.join()

    print(‘主线程‘)
    
# 运行结果
子进程1 is running
子进程2 is running
3
子进程2 is done
子进程1 is done
主线程

eumecate()把当前活跃线程对象拿出来

from threading import Thread,current_thread,active_count,enumerate
import time,os

def task(name):
    print(‘%s is running‘%name)
    time.sleep(2)
    print(‘%s is done‘%name)

if __name__ == ‘__main__‘:
    t1 = Thread(target=task,args=(‘子进程1‘,))
    t2 = Thread(target=task,args=(‘子进程2‘,))
    t1.start()
    t2.start()
    print(‘当前线程存活数目：‘,active_count())
    print(‘当前活跃线程对象：‘,enumerate())
    print(‘主线程‘)
    
# 运行结果：
子进程1 is running
子进程2 is running
当前线程存活数目： 3
当前活跃线程对象： [<_MainThread(MainThread, started 14472)>, <Thread(Thread-1, started 8036)>, <Thread(Thread-2, started 12328)>]
主线程
子进程1 is done
子进程2 is done

守护线程

无论是进程还是线程，都遵循：守护xx会等待主xx运行完毕后被销毁

需要强调的是，运行完毕并非是终止运行

1.对主进程来说，运行完毕指的是主进程代码运行完毕
2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕后，主线程才算是运行完毕

详细解释：

1.主进程在其代码结束后就已经算是运行完毕了（守护进程此时就会被回收掉），然后主进程会一直等非守护进程的子进程都运行完毕后回收子进程的资源（否则会产生僵尸进程），才会结束。

2.主线程在其他非守护线程运行完毕后才算运行完毕（守护进程在此时就会回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都完毕后才能结束。

代码演示

# 只有一个线程

from threading import Thread
import time

def sayhi(name):
    time.sleep(2)
    print(‘%s is say hello‘%name)

if __name__ == ‘__main__‘:
    t = Thread(target=sayhi,args=(‘肖亚飞‘))
    # t.setDaemon(True)  # 设置成为守护线程
    t.daemon = True
    t.start()
    print(‘主线程‘)  # 到此时主线程运行完毕
    print(t.is_alive())  # 判断线程是否存活
    
# 运行结果为：
主线程
True

# 有多个线程

from threading import Thread
import time

def foo():
    print(123)
    time.sleep(2)
    print(‘end123‘)

def bar():
    print(456)
    time.sleep(2)
    print(‘end456‘)

if __name__ == ‘__main__‘:
    t1 = Thread(target=foo,)
    t2 = Thread(target=bar,)

    t1.daemon = True  # 将t1设置成为守护线程
    t1.start()
    t2.start()
    print(‘主线程‘)  # 主线程代码
    
# 运行结果为：
123
456
主线程
end123
end456

死锁现象与递归锁

死锁现象

所谓死锁：两个或两个以上的进程/线程在执行过程中，因争夺资源而造成的一种互相等待的现象，苦无外力作用，他们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程成为死锁进程，如下就是死锁：

# 死锁
from threading import Thread,Lock
import time

# 创建A锁和B锁
mutexA = Lock()  # 互斥锁只能acquire一次
mutexB = Lock()

class MyThread(Thread):

    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print(‘%s 拿到了A锁‘%self.name)

        mutexB.acquire()
        print(‘%s 拿到了B锁‘%self.name)
        mutexB.release()

        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print(‘%s 拿到了B锁‘%self.name)
        time.sleep(0.05)

        mutexA.acquire()
        print(‘%s 拿到了A锁‘%self.name)
        mutexA.release()

        mutexB.release()

if __name__ == ‘__main__‘:
    for i in range(10):
        t = MyThread()
        t.start()  # 执行run方法
        
# 运行结果为：
Thread-1 拿到了A锁
Thread-1 拿到了B锁
Thread-1 拿到了B锁
Thread-2 拿到了A锁

说明：在这段代码中，f2 方法为首先拿到mutexB这把锁，在拿到之后会沉睡0.05s，操作系统会觉得这是一个IO堵塞，然后将COU切换给Thread-2使用，然后Thread-2拿到了mutexA锁后，发现：进入了死锁

而解决死锁的方法就是使用递归锁

递归锁

我们刚刚见过，互斥锁只能acquire一次，但是递归锁就可以acquire多次，就相当于每acquire一次，它的计数器就会增加1，每release一次就是减少1，当计数为0时，才可以被其它线程抢到acquire。

# 使用递归锁解决死锁问题
from threading import Thread,RLock
import time

# 创建锁
mutexA = RLock()
mutexB = mutexA  # 相当于一把锁

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print(‘%s 拿到了A锁‘%self.name)

        mutexB.acquire()
        print(‘%s 拿到了B锁‘%self.name)
        mutexB.release()

        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print(‘%s 拿到了B锁‘%self.name)
        time.sleep(1)

        mutexA.acquire()
        print(‘%s 拿到了A锁‘%self.name)
        mutexA.release()

        mutexB.release()

if __name__ == ‘__main__‘:
    for i in range(10):
        t = MyThread()
        t.start()

信号量

互斥锁就是在同一时间只能有一个任务抢到锁去执行，而信号量也是一把锁，可以指定信号量为5，信号量就是同一时间有5个任务拿到锁去执行，如果说互斥锁是合租房屋里的人去抢一个厕所，那么信号量就相当于一群路人争夺公共厕所，公共厕所的人数有限制，这就意味着同一时间可以有多少人上公共厕所，但公共厕所容纳的人数也是一定的，这就是信号量的大小。

# 信号量
from threading import Thread,Semaphore,current_thread
import time,random

sm = Semaphore(3)  # 设置信号量大小为3

def task():
    sm.acquire()
    print(‘%s in‘%current_thread().getName())
    time.sleep(random.randint(1,3))
    sm.release()

if __name__ == ‘__main__‘:
    for i in range(10):
        t = Thread(target=task,)
        t.start()
      
# 运行结果为
Thread-1 in
Thread-2 in
Thread-3 in

Thread-4 in

Thread-5 in

Thread-6 in
Thread-7 in
Thread-8 in

Thread-9 in

Thread-10 in


# 其中，拿到锁和释放锁还可以有这么种写法
    with sm:
        print(‘%s in ‘%current_thread().getName())
        time.sleep(random.randint(1,3))

解析

Semaphore管理一个内置的计数器
每当调用acquire()时内置计数器会-1
调用release()时内置计数器+1
计数器不能小于0；当计数器为0时，acquire()将阻塞线程知道其他线程调用release()

Event定时器

线程的一个关键特性是每个线程都是独立运行且状态不可预测的。如果程序中的其他线程需要通过判断某个线程的状态来确定自己的下一步操作，这时线程同步问题就会变得很棘手。为了解决这些问题，我们需要使用threading库中的Event对象。对象包含一个可由线程设置的信号标志，它允许线程等到某些事情的发生。在初始情况下，Event对象中的信号标志被设置成为False，如果有线程等待Event对象，而这个Event对象的标志为假，那么这个线程就会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真，它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置成为真的Event对象，那么它将忽略这个事情，继续执行。

from Threading import Event
event = Event()

event.isSet()  # 返回event的状态值

event.wait()  # 如果event.isSet == False，则阻塞线程

event.set()  # 设置event的状态值为True，所有阻塞线程激活进入就绪状态，等到操作系统调度

event.clear()  # 恢复event的状态值为True

在高中上课时，老师讲课学生听课，学生想要下课必须等待老师发送下课信号

from threading import Thread,Event
import time

event = Event()

def student(name):
    print(‘%s 正在听课‘%name)
    event.wait()
    print(‘%s 课件活动‘%name)

def teacher(name):
    print(‘老师 %s 正在讲课 ‘%name)
    time.sleep(7)
    event.set()

if __name__ == ‘__main__‘:
    stu1 = Thread(target=student,args=(‘李鹏‘,))
    stu2 = Thread(target=student,args=(‘李坤‘,))
    stu3 = Thread(target=student,args=(‘魏文武‘,))
    t1 = Thread(target=teacher,args=(‘肖亚飞‘,))

    stu1.start()
    stu2.start()
    stu3.start()
    t1.start()

# 运行结果
李鹏 正在听课
李坤 正在听课
魏文武 正在听课
老师 肖亚飞 正在讲课 

李坤 课件活动
魏文武 课件活动
李鹏 课件活动

大学生活，老师讲他自己的，学生想下课就下课

from threading import Thread,Event
import time
event = Event()

def student(name):
    print(‘学生 <%s> 正在听课‘%name)
    event.wait(3)  # 设置超时时间，等待2秒就走
    print(‘学生 <%s> 课件活动‘%name)
def teacher(name):
    print(‘%s 正在讲课‘%name)
    time.sleep(7)
    event.set()

if __name__ == ‘__main__‘:
    stu1 = Thread(target=student,args=(‘李鹏‘,))
    stu2 = Thread(target=student,args=(‘李坤‘,))
    stu3 = Thread(target=student,args=(‘大山‘,))
    t1 = Thread(target=teacher,args=(‘肖亚飞‘,))

    stu1.start()
    stu2.start()
    stu3.start()
    t1.start()
    
# 运行结果
学生 <李鹏> 正在听课
学生 <李坤> 正在听课
学生 <大山> 正在听课
肖亚飞 正在讲课

学生 <李坤> 课件活动    # 3s超时时间过去了
学生 <李鹏> 课件活动
学生 <大山> 课件活动

例如，有多个工作线程尝试连接MySQL，我们想要在连接前确保MySQL服务正常才能让那些工作线程去连接MySQL服务器，如果连接不成功，都回去尝试连接。那么我们需要采用Event机制来协调各个线程之间的连接操作

# 检查mysql连接
from threading import Thread,Event,current_thread
import time,random

# 先生成定时器对象
event = Event()

def conn_mysql():
    count = 1
    while not event.is_set():  # 默认为False
        if count > 3:
            raise TimeoutError(‘连接超时‘)
        print(‘<%s> 第%s次尝试连接‘%(current_thread().getName(),count))
        event.wait(0.5)
        count += 1
        print(‘<%s> 连接成功‘%current_thread().getName())

def check_mysql():
    print(‘%s is checking‘%current_thread().getName())
    time.sleep(5)
    event.set()

if __name__ == ‘__main__‘:
    for i in range(3):
        t = Thread(target=conn_mysql,)
        t.start()
    t = Thread(target=check_mysql)
    t.start()

# 运行结果为
<Thread-1> 第1次尝试连接
<Thread-2> 第1次尝试连接
<Thread-3> 第1次尝试连接
Thread-4 is checking
<Thread-1> 连接成功
<Thread-2> 连接成功
<Thread-3> 连接成功
<Thread-3> 第2次尝试连接
<Thread-1> 第2次尝试连接
<Thread-2> 第2次尝试连接
<Thread-2> 连接成功
<Thread-1> 连接成功
<Thread-1> 第3次尝试连接
<Thread-3> 连接成功
<Thread-2> 第3次尝试连接
<Thread-3> 第3次尝试连接
<Thread-2> 连接成功
<Thread-1> 连接成功
<Thread-3> 连接成功

定时器

定时器，指定n秒后执行某操作

# 定时器
from threading import Timer
def hello():
    print(‘hello world‘)

t = Timer(2,hello)
t.start()

基于定时器实现验证码登录

# 基于定时器实现验证码登录
from threading import Thread,Timer
import random

class Code():
    def __init__(self):
        self.make_cache()

    # 在登录的时候就应该有一个验证码
    def make_cache(self,interval=5):
        self.cache = self.make_code()
        print(self.cache)
        self.t = Timer(interval,self.make_cache)
        self.t.start()

    # 生成验证码
    def make_code(self,n=4):  # 验证码的个数
         res = ‘‘
         for i in range(n):
             s1 = str(random.randint(0,9))
             s2 = chr(random.randint(65,90))
             res += random.choice([s1,s2])  # 随机字符串
         return res

    # 验证
    def check(self):
        while True:
            code = input(‘请输入验证码>>>‘).strip()
            if code.upper() == self.cache:
                print(‘验证码输入正确‘)
                self.t.cancel()
                break

obj = Code()
obj.check()

# 运行结果为
IFI4
请输入验证码>>>PRC0
PRCO878M

请输入验证码>>>878M
验证码输入正确

## 线程Queue

在线程编程中，当信息必须在多个线程之间安全地交换时，队列特别有用。

线程有三种用法

• 队列：先进先出
• 堆栈：先进后出
• 优先级队列：存储数据时可设置优先级的队列

下面依次来演示一下

队列：先进先出

# 线程queue-->队列
import queue
q = queue.Queue(3)  # 设置队列为3
q.put(‘first‘)
q.put(2)
q.put(‘third‘)

# 打印
print(q.get())
print(q.get())
print(q.get())

# 打印结果为
first
2
third

堆栈：先进后出

# 线程queue-->堆栈
import queue
q = queue.LifoQueue(3)
q.put(1)
q.put(2)
q.put(3)

print(q.get())
print(q.get())
print(q.get())

# 打印结果为
3
2
1

优先级队列：存储数据时可设置优先级的队列

# 线程queue-->优先级
import queue
q = queue.PriorityQueue(3)
q.put((10,‘first‘))  # 10 代表优先级
q.put((40,2))
q.put((20,‘third‘))

print(q.get())  # 数字越小优先级越高
print(q.get())
print(q.get())

# 打印结果为
(10, ‘first‘)
(20, ‘third‘)
(40, 2)

进程池和线程池

在刚开始学多线程或多进程时，我们迫不及待的基于多线程、多进程实现并发的套接字通信，然而这种实现方式的致命缺陷是：服务的开启的进程数或线程数会随着并发的客户端的数目增多而增多，这会对服务器主机带来巨大的压力，甚至瘫痪，于是我们必须对服务器开启的进程数或线程数加以控制，让机器在一个自己能够承受的范围内运行，这就是进程池和线程池的用途，例如进程池，就是用来存放进程的池子，本质上还是基于多进程，只不过是对开启进程的数目加上了限制

介绍

官网：https://docs.python.org/dev/library/concurrent.futures.html

concurrent.futures  模块提供了高度封装的异步调用接口
ThreadPoolExecutor   线程池，提供异步调用
ProcessPoolExecutor  进程池，提供异步调用

Both implement the same interface, which is defined by the abstract Executor class.两者实现相同的接口，抽象Executor类定义该接口。

基本方法

1.submit(fn,*args,**kwargs)
异步提交任务

2.map(func,*iterables,timeout=None,chunksize=1)
取代for循环submit操作

3.shutdown(wait=True)
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕后收回资源才继续
wait=False，立即返回，并不会原地等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

4.result(timeout=None)
取到结果
 
5.add_done_callbakc(n)
回调函数

进程池

用法

from concurrent.futures import ProcessPoolExecutor
import os,time, random

def task(name):
    print(‘name:%s  pid:%s‘%(name,os.getpid()))
    time.sleep(random.randint(1,3))

if __name__ == ‘__main__‘:
    pool = ProcessPoolExecutor(4)  # 如果不指定的话，则为本机cpu的核数
    for i in range(10):
        pool.submit(task,‘肖亚飞%s‘%i)  # 异步提交：提交完任务后不用在原地等待

    # 维持计数器，一共有10个任务，运行一个就会少1个
    pool.shutdown(wait=True)
    print(‘主进程‘)

那么把这段代码运行了之后，就可以很明显的看见有效的PID就只有4个，从而很好的控制了PID

线程池

用法

from concurrent.futures import ThreadPoolExecutor
import os,time,random

def task(name):
    print(‘name:%s pid:%s‘%(name,os.getpid()))
    time.sleep(random.randint(0.01-0.02))

if __name__ == ‘__main__‘:
    start_time = time.time()
    pool = ThreadPoolExecutor(4)
    for i in range(10000):
        pool.submit(task,‘肖亚飞%s‘%i)

    pool.shutdown(wait=True)
    stop_time = time.time()
    print(‘主线程‘,stop_time-start_time)

查看线程的名字

# 查看线程的名字
from concurrent.futures import  ThreadPoolExecutor
import time,random,os
from threading import current_thread

def task():
    print(‘name:%s pid:%s‘%(current_thread().getName(),os.getpid()))
    time.sleep(random.randint(1,3))

if __name__ == ‘__main__‘:
    pool = ThreadPoolExecutor(max_workers=4)
    for i in range(100):
        pool.submit(task,)
    pool.shutdown(wait=True)
    print(‘主进程‘)

map方法

map(func,*iterables,timeout=None,chunksize=1) 取代for循环submit操作

# map方法
from concurrent.futures import ThreadPoolExecutor
import time,os
from threading import current_thread

def task():
    print(‘name:%s pid:%s‘%(current_thread().getName(),os.getpid()))
    time.sleep(1)

if __name__ == ‘__main__‘:
    pool = ThreadPoolExecutor(max_workers=4)
    pool.map(task,range(1,12))

回调函数

可以为进程池或线程池内的每个进程或线程绑定一个函数，该函数在进程或线程任务执行完毕后自动触发，并接受任务的返回值当做参数，该函数成为回调函数

from concurrent.futures import ThreadPoolExecutor
import requests
import os

def get_page(url):
    print(‘<进程%s> get %s‘ %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {‘url‘:url,‘text‘:respone.text}

def parse_page(res):
    res=res.result()
    print(‘<进程%s> parse %s‘ %(os.getpid(),res[‘url‘]))
    parse_res=‘url:<%s> size:[%s]
‘ %(res[‘url‘],len(res[‘text‘]))
    with open(‘db.txt‘,‘a‘) as f:
        f.write(parse_res)


if __name__ == ‘__main__‘:
    urls=[
        ‘https://www.baidu.com‘,
        ‘https://www.python.org‘,
        ‘https://www.openstack.org‘,
        ‘https://help.github.com/‘,
        ‘http://www.sina.com.cn/‘
    ]

    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page)  # parse_page拿到的是一个futuer对象，需要obj,result()拿到结果

 # 运行结果为
<进程2392> get https://www.baidu.com
<进程13288> get https://www.python.org
<进程1268> get https://www.openstack.org
<进程2392> get https://help.github.com/
<进程13984> parse https://www.baidu.com
<进程1268> get http://www.sina.com.cn/
<进程13984> parse https://www.openstack.org
<进程13984> parse http://www.sina.com.cn/
<进程13984> parse https://www.python.org
<进程13984> parse https://help.github.com/

提交任务的两种方式

• 同步提交：也就是提交完任务后，就在原地等待任务执行完毕，拿到结果再去执行下一段代码
• 异步提交：提交完任务后不在原地等待执行完毕

我们来模拟一下同步提交和异步提交

# 同步提交
from concurrent.futures import ThreadPoolExecutor
import time,random

def la(name):
    print(‘%s is laing‘%name)
    time.sleep(random.randint(3,5))
    res = random.randint(7,13)*‘#‘
    return {‘name‘:name,‘res‘:res}

def weight(shit):
    name = shit[‘name‘]
    size = len(shit[‘res‘])
    print(‘%s 拉了 %s kg‘%(name,size))

if __name__ == ‘__main__‘:
    pool = ThreadPoolExecutor(13)
    shit1 = pool.submit(la,‘李鹏‘).result()
    weight(shit1)
    shit2 = pool.submit(la,‘李坤‘).result()
    weight(shit2)
    shit3 = pool.submit(la,‘魏文武‘).result()
    weight(shit3)
    
# 运行结果
李鹏 is laing
李鹏 拉了 9 kg
李坤 is laing
李坤 拉了 7 kg
魏文武 is laing
魏文武 拉了 9 kg

# 异步调用
from concurrent.futures import ThreadPoolExecutor
import time,random

def la(name):
    print(‘%s is laing‘%name)
    time.sleep(random.randint(3,5))
    res = random.randint(7,13)*‘#‘
    return {‘name‘:name,‘res‘:res}

def weight(shit):
    shit = shit.result()  # 拿到这个结果
    name = shit[‘name‘]
    res = len(shit[‘res‘])
    print(‘%s 拉了 %s kg‘%(name,res))

if __name__ == ‘__main__‘:
    pool = ThreadPoolExecutor(13)
    pool.submit(la,‘李鹏‘).add_done_callback(weight)  # 绑定回调函数，前面一个任务执行完成后，return返回的值（futuer对象）就会当做参数传递个weight函数
    pool.submit(la,‘李坤‘).add_done_callback(weight)
    pool.submit(la,‘魏文武‘).add_done_callback(weight)

# 运行结果
李鹏 is laing
李坤 is laing
魏文武 is laing
魏文武 拉了 12 kg
李坤 拉了 12 kg
李鹏 拉了 11 kg