1、并发与并行
并行：多个程序同时运行
并发：伪并行，看起来是同时并行，其实质是利用了多道技术
无论是并行还是并发，在用户眼里看起来都是同时运行的，不管是线程还是进程，都是只是一个任务，真正干活的是CPU，而同一个CPU在同一时刻只能执行一个任务。
2、多进程
- 进程是操作系统分配资源的最小单元，线程是操作系统调度的最小单元
- 一个应用程序最少包括一个进程，而一个进程包括1个或多个线程，线程的尺度更小
- 每个进程在执行过程中都有独立的内存单元，而每一个进程中的多个线程在执行过程中共享内存
2.1 进程之 Process
python的多进程编程主要依靠multiprocess模块。我们先对比两段代码，看看多进程编程的优势。我们模拟了一个非常耗时的任务，计算8的20次方，为了使这个任务显得更耗时，我们还让它sleep 2秒。第一段代码是单进程计算(代码如下所示)，我们按顺序执行代码，重复计算2次，并打印出总共耗时。
# -*- coding: utf-8 -*-
# Created by Xue Jian on 3/10/23
import time
import os
def long_time_task():
print('current_process: {}'.format(os.getpid()))
time.sleep(2)
print('result: {}'.format(8 ** 20))
if __name__ == '__main__':
print('current parent process: {}'.format(os.getpid()))
start = time.time()
for i in range(2):
long_time_task()
end = time.time()
print('sec: {}'.format(end - start))
输出结果如下：
总耗时4秒，自始至终只有一个进程在执行。看来电脑计算8的20次方基本不费时。
第2段代码是多进程计算代码。我们利用multiprocess模块的Process方法创建了两个新的进程p1和p2来进行并行计算。Process方法接收两个参数, 第一个是target，一般指向函数名，第二个时args，需要向函数传递的参数。对于创建的新进程，调用start()方法即可让其开始。我们可以使用os.getpid()打印出当前进程的名字。
# -*- coding: utf-8 -*-
# Created by Xue Jian on 3/10/23
import time
import os
from multiprocessing import Process
def long_time_task(i):
print('子进程: {} - 任务{}'.format(os.getpid(), i))
time.sleep(2)
print("结果: {}".format(8 ** 20))
if __name__ == '__main__':
print('当前母进程: {}'.format(os.getpid()))
start = time.time()
p1 = Process(target=long_time_task, args=(1,))
p2 = Process(target=long_time_task, args=(2,))
print('等待所有子进程完成。')
p1.start()
p2.start()
p1.join()
p2.join()
end = time.time()
print("总共用时{}秒".format((end - start)))
输出结果如下所示，耗时变为2秒，时间减了一半，可见并发执行的时间明显比顺序执行要快很多。你还可以看到尽管我们只创建了两个进程，可实际运行中却包含里1个母进程和2个子进程。之所以我们使用join()方法就是为了让母进程阻塞，等待子进程都完成后才打印出总共耗时，否则输出时间只是母进程执行的时间。
知识点：
- 新创建的进程与进程的切换都是需要消耗资源的，所以平时工作中进程数不能开太大
- 同时可以运行的进程数，一般受制于CPU的核数
- 除了使用Process方法，我们还可以使用Pool方法创建多进程
2.2 进程之Pool类
很多时候系统都需要创建多个进程以提高CPU的利用率，当数量较少时，可以手动生成一个个Process实例。
当进程数量很多时，或许可以利用循环，但是这需要程序员手动管理系统中并发进程的数量，有时会很麻烦。这时进程池Pool就可以发挥其功效了。可以通过传递参数限制并发进程的数量，默认值为CPU的核数。
Pool类可以提供指定数量的进程供用户调用，当有新的请求提交到Pool中时，如果进程池还没有满，就会创建一个新的进程来执行请求。如果池满，请求就会告知先等待，直到池中有进程结束，才会创建新的进程来执行这些请求。
下面介绍一下multiprocessing 模块下的Pool类的几个方法：
1.apply_async
函数原型：apply_async(func[, args=()[, kwds={}[, callback=None]]])
其作用是向进程池提交需要执行的函数及参数， 各个进程采用非阻塞（异步）的调用方式，即每个子进程只管运行自己的，不管其它进程是否已经完成。这是默认方式。
2.map()
函数原型：map(func, iterable[, chunksize=None])
Pool类中的map方法，与内置的map函数用法行为基本一致，它会使进程阻塞直到结果返回。 注意：虽然第二个参数是一个迭代器，但在实际使用中，必须在整个队列都就绪后，程序才会运行子进程。
3.map_async()
函数原型：map_async(func, iterable[, chunksize[, callback]])
与map用法一致，但是它是非阻塞的。其有关事项见apply_async。
4.close()
关闭进程池（pool），使其不在接受新的任务。
5. terminate()
结束工作进程，不在处理未处理的任务。
6.join()
主进程阻塞等待子进程的退出， join方法要在close或terminate之后使用。
下例是一个简单的multiprocessing.Pool类的实例。
进程池会根据我的CPU的核数进行，本次使用的虚拟机是4核的，这里会创建一个容量为4的进程池，4个进程需要执行5个任务，会有一个在排队等待
from multiprocessing import Pool, cpu_count, Process
import os
import time
def long_time_task(i):
print('子进程: {} - 任务{}'.format(os.getpid(), i))
time.sleep(2)
print("结果: {}".format(8 ** 20))
if __name__ == '__main__':
print("CPU内核数:{}".format(cpu_count()))
print('当前母进程: {}'.format(os.getpid()))
start = time.time()
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print('等待所有子进程完成。')
p.close()
p.join()
end = time.time()
print("总共用时{}秒".format((end - start)))
2.3 多进程之间的数据共享和通信
通常，进程之间是相互独立的，每个进程都有独立的内存。通过共享内存(nmap模块)，进程之间可以共享对象，使多个进程可以访问同一个变量(地址相同，变量名可能不同)。
多进程共享资源必然会导致进程间相互竞争，所以应该尽最大可能防止使用共享状态。还有一种方式就是使用队列queue来实现不同进程间的通信或数据共享，这一点和多线程编程类似。
下例这段代码中中创建了2个独立进程，一个负责写(pw), 一个负责读(pr), 实现了共享一个队列queue
from multiprocessing import Process, Queue
import os, time, random
# 写数据进程执行的代码:
def write(q):
print('Process to write: {}'.format(os.getpid()))
for value in ['A', 'B', 'C']:
print('Put %s to queue...' % value)
q.put(value)
time.sleep(random.random())
# 读数据进程执行的代码:
def read(q):
print('Process to read:{}'.format(os.getpid()))
while True:
value = q.get(True)
print('Get %s from queue.' % value)
if __name__ == '__main__':
# 父进程创建Queue，并传给各个子进程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 启动子进程pw，写入:
pw.start()
# 启动子进程pr，读取:
pr.start()
# 等待pw结束:
pw.join()
# pr进程里是死循环，无法等待其结束，只能强行终止:
pr.terminate()
3. 线程
python3 中的多线程主要是依靠 threading模块，创建新线程与创建新进程的方法非常类似。
threading.Thread方法可以接收两个参数, 第一个是target，一般指向函数名，第二个是args，需要向函数传递的参数。
对于创建的新线程，调用start()方法即可让其开始。我们还可以使用current_thread().name打印出当前线程的名字。 下例中我们使用多线程技术重构之前的计算代码
import threading
import time
def long_time_task(i):
print('当前子线程: {} - 任务{}'.format(threading.current_thread().name, i))
time.sleep(2)
print("结果: {}".format(8 ** 20))
if __name__=='__main__':
start = time.time()
print('这是主线程：{}'.format(threading.current_thread().name))
t1 = threading.Thread(target=long_time_task, args=(1,))
t2 = threading.Thread(target=long_time_task, args=(2,))
t1.start()
t2.start()
end = time.time()
print("总共用时{}秒".format((end - start)))
当我们设置多线程时，主线程会创建多个子线程，在python中，默认情况下主线程和子线程独立运行互不干涉。
如果希望让主线程等待子线程实现线程的同步，我们需要使用join()方法。
如果我们希望一个主线程结束时不再执行子线程，我们应该怎么办呢? 我们可以使用t.setDaemon(True)，代码如下所示。
import threading
import time
def long_time_task():
print('当子线程: {}'.format(threading.current_thread().name))
time.sleep(2)
print("结果: {}".format(8 ** 20))
if __name__ == '__main__':
start = time.time()
print('这是主线程：{}'.format(threading.current_thread().name))
for i in range(5):
t = threading.Thread(target=long_time_task, args=())
t.setDaemon(True)
t.start()
end = time.time()
print("总共用时{}秒".format((end - start)))
以上可以看到，当主线程执行完成时，并不会等待子线程执行完成，而是直接完成
3.1 通过继承Thread类重写run方法创建新进程
除了使用Thread()方法创建新的线程外，我们还可以通过继承Thread类重写run方法创建新的线程，这种方法更灵活。下例中我们自定义的类为MyThread, 随后我们通过该类的实例化创建了2个子线程。
import threading
import time
def long_time_task(i):
time.sleep(2)
return 8 ** 20
class MyThread(threading.Thread):
def __init__(self, func, args, name='', ):
threading.Thread.__init__(self)
self.func = func
self.args = args
self.name = name
self.result = None
def run(self):
print('开始子进程{}'.format(self.name))
self.result = self.func(self.args[0], )
print("结果: {}".format(self.result))
print('结束子进程{}'.format(self.name))
if __name__ == '__main__':
start = time.time()
threads = []
for i in range(1, 3):
t = MyThread(long_time_task, (i,), str(i))
threads.append(t)
for t in threads:
t.start()
for t in threads:
t.join()
end = time.time()
print("总共用时{}秒".format((end - start)))
3.2 不同线程间的数据共享
一个进程中的不同线程之间是共享内存的，这就意味着任何一个变量都可以被任何一个线程修改，因此线程之间共享数据最大的危险在于多个线程同时修改同一个变量，把内容给改乱了。
如果不同线程之间有共享的变量，其中一个方法就是在修改前给加上一把锁lock，确保一次只有一个线程能修改它。
threading.Lock()方法可以轻易实现对一个共享变量的锁定，修改后release，以供其他线程使用。
比如在下例中 余额 balance是一个共享变量，使用lock可以使其不被改变
import threading
class Account:
def __init__(self):
self.balance = 0
def add(self, lock):
# 获得锁
# lock.acquire()
for i in range(0, 100000):
self.balance += 1
# 释放锁
# lock.release()
def delete(self, lock):
# 获得锁
# lock.acquire()
for i in range(0, 100000):
self.balance -= 1
# 释放锁
# lock.release()
if __name__ == "__main__":
account = Account()
lock = threading.Lock()
# 创建线程
thread_add = threading.Thread(target=account.add, args=(lock,), name='Add')
thread_delete = threading.Thread(target=account.delete, args=(lock,), name='Delete')
# 启动线程
thread_add.start()
thread_delete.start()
# 等待线程结束
thread_add.join()
thread_delete.join()
print('The final balance is: {}'.format(account.balance))
上面的代码没有对balance变量进行加锁，所以在操作过程中会导致异常情况出现。
下图进行加锁，就可以避免这个情况
import threading
class Account:
def __init__(self):
self.balance = 0
def add(self, lock):
# 获得锁
lock.acquire()
for i in range(0, 100000):
self.balance += 1
# 释放锁
lock.release()
def delete(self, lock):
# 获得锁
lock.acquire()
for i in range(0, 100000):
self.balance -= 1
# 释放锁
lock.release()
if __name__ == "__main__":
account = Account()
lock = threading.Lock()
# 创建线程
thread_add = threading.Thread(target=account.add, args=(lock,), name='Add')
thread_delete = threading.Thread(target=account.delete, args=(lock,), name='Delete')
# 启动线程
thread_add.start()
thread_delete.start()
# 等待线程结束
thread_add.join()
thread_delete.join()
print('The final balance is: {}'.format(account.balance))
另一种实现不同线程间数据共享的方法就是使用消息队列queue。不像列表，queue是线程安全的，可以放心使用
使用queue队列通信-经典的生产者和消费者模型
下例创建两个线程，一个负责生产，一个负责消费，所生产的产品存放在queue里，实现了不同线程的沟通
from queue import Queue
import random, threading, time
# 生产者类
class Producer(threading.Thread):
def __init__(self, name, queue):
threading.Thread.__init__(self, name=name)
self.queue = queue
def run(self):
for i in range(1, 5):
print("{} is producing {} to the queue!".format(self.getName(), i))
self.queue.put(i)
time.sleep(random.randrange(10) / 5)
print("%s finished!" % self.getName())
# 消费者类
class Consumer(threading.Thread):
def __init__(self, name, queue):
threading.Thread.__init__(self, name=name)
self.queue = queue
def run(self):
for i in range(1, 5):
val = self.queue.get()
print("{} is consuming {} in the queue.".format(self.getName(), val))
time.sleep(random.randrange(10))
print("%s finished!" % self.getName())
def main():
queue = Queue()
producer = Producer('Producer', queue)
consumer = Consumer('Consumer', queue)
producer.start()
consumer.start()
producer.join()
consumer.join()
print('All threads finished!')
if __name__ == '__main__':
main()
队列queue的put方法可以将一个对象obj放入队列中。如果队列已满，此方法将阻塞至队列有空间可用为止。queue的get方法一次返回队列中的一个成员。如果队列为空，此方法将阻塞至队列中有成员可用为止。queue同时还自带emtpy(), full()等方法来判断一个队列是否为空或已满，但是这些方法并不可靠，因为多线程和多进程，在返回结果和使用结果之间，队列中可能添加/删除了成员。
4. Python多进程和多线程哪个快?
由于GIL的存在。很多人认为python多进程编程更快，针对多核CPU，理论上来说也是采用多进程更能有效利用资源。
- 对于CPU密集型代码(比如循环计算) -- 多进程效率更高
- 对于IO密集型代码(比如文件操作，网络爬虫) -- 多线程效率更高
5. 线程池
Python标准库为我们提供了threading和multiprocessing模块编写相应的多线程/多进程代码，但是当项目达到一定的规模，频繁创建/销毁进程或者线程是非常消耗资源的，这个时候我们就要编写自己的线程池/进程池，以空间换时间。
但从Python3.2开始，标准库为我们提供了concurrent.futures模块，它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，实现了对threading和multiprocessing的进一步抽象，对编写线程池/进程池提供了直接的支持。
5.1 参数详解
ProcessPoolExecutor(n):n表示池里面存放多少个进程，之后的连接最大就是n的值 submit(fn,*args,**kwargs) 异步提交任务 map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操作 shutdown(wait=True) 相当于进程池的pool.close()+pool.join()操作 wait=True，等待池内所有任务执行完毕回收完资源后才继续，--------》默认 wait=False，立即返回，并不会等待池内的任务执行完毕 但不管wait参数为何值，整个程序都会等到所有任务执行完毕 submit和map必须在shutdown之前 result(timeout=None) #取得结果 add_done_callback(fn) #回调函数
使用submit 来操作线程池/进程池
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import time
# 模拟网络请求的网络延迟
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
# 创建一个大小为2的线程池
pool = ThreadPoolExecutor(max_workers=2)
# 将上个任务提交到线程池，因为线程池的大小是2，所以必须等task1和task2中有一个完成之后才会将第三个任务提交到线程池
task1 = pool.submit(get_html, 3)
task2 = pool.submit(get_html, 2)
task3 = pool.submit(get_html, 4)
# 打印该任务是否执行完毕
print(task1.done())
# 只有未被提交的到线程池（在等待提交的队列中）的任务才能够取消
print(task3.cancel())
time.sleep(4) # 休眠4秒钟之后，线程池中的任务全部执行完毕，可以打印状态
print(task1.done())
print(task1.result()) # 该任务的return 返回值 该方法是阻塞的。
- ThreadPoolExecutor构造实例的时候，传入max_workers参数来设置线程池中最多能同时运行的线程数目。
- 使用submit函数来提交线程需要执行的任务（函数名和参数）到线程池中，并返回该任务的句柄（类似于文件、画图），注意submit()不是阻塞的，而是立即返回。
- 通过submit函数返回的任务句柄，能够使用done()方法判断该任务是否结束。上面的例子可以看出，由于任务有2s的延时，在task1提交后立刻判断，task1还未完成，而在延时4s之后判断，task1就完成了。
- 使用cancel()方法可以取消提交的任务，如果任务已经在线程池中运行了，就取消不了。这个例子中，线程池的大小设置为2，任务已经在运行了，所以取消失败。如果改变线程池的大小为1，那么先提交的是task1，task2还在排队等候，这是时候就可以成功取消。
- 使用result()方法可以获取任务的返回值。查看内部代码，发现这个方法是阻塞的。
as_completed
上面虽然提供了判断任务是否结束的方法，但是不能在主线程中一直判断啊。有时候我们是得知某个任务结束了，就去获取结果，而不是一直判断每个任务有没有结束。这是就可以使用as_completed方法一次取出所有任务的结果。
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
# 模拟网络请求的网络延迟
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
pool = ThreadPoolExecutor(max_workers=2)
urls = [2, 3, 4]
all_task = [pool.submit(get_html, url) for url in urls]
for future in as_completed(all_task):
data = future.result()
print("in main: get page {}s success".format(data))
as_completed()方法是一个生成器，在没有任务完成的时候，会阻塞，在有某个任务完成的时候，会yield这个任务，就能执行for循环下面的语句，然后继续阻塞住，循环到所有的任务结束。从结果也可以看出，先完成的任务会先通知主线程。
map
除了上面的as_completed方法，还可以使用executor.map方法，但是有一点不同。
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
# 模拟网络请求的网络延迟
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
pool = ThreadPoolExecutor(max_workers=2)
urls = [2, 3, 4]
for data in pool.map(get_html, urls):
print("in main: get page {}s success".format(data))
shutdown
shutdown方法的功能类似于 join+close的
import time
from concurrent.futures import ThreadPoolExecutor, wait
results = []
def get_html(i):
time.sleep(2)
return 2 * i
def handle(res):
res = res.result()
results.append(res)
pool = ThreadPoolExecutor(max_workers=2)
for i in range(4):
pool.submit(get_html, i).add_done_callback(handle)
pool.shutdown(wait=True) # 相当于 join + close
print('main')
print(results)
wait
wait方法可以让主线程阻塞，直到满足设定的要求。
当设置了wait后，主线程会一直等待子线程执行完毕才能执行
和 shutdown效果类似
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time
# 参数times用来模拟网络请求的时间
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
6. 进程池
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time,random,os
def task(n):
print('%s is running'% os.getpid())
time.sleep(random.randint(1,3))
return n
def handle(res):
res=res.result()
print("handle res %s"%res)
if __name__ == '__main__':
#同步调用
# pool=ProcessPoolExecutor(8)
#
# for i in range(13):
# pool.submit(task, i).result() #变成同步调用，串行了，等待结果
# # pool.shutdown(wait=True) #关门等待所有进程完成
# pool.shutdown(wait=False)#默认wait就等于True
# # pool.submit(task,3333) #shutdown后不能使用submit命令
#
# print('主')
#异步调用
pool=ProcessPoolExecutor(8)
for i in range(13):
obj=pool.submit(task,i)
obj.add_done_callback(handle) #这里用到了回调函数
pool.shutdown(wait=True) #关门等待所有进程完成
print('主')
##注意，创建进程池必须在if __name__ == '__main__':中，否则会报错
##其他的用法和创建线程池的一样
from concurrent.futures import ThreadPoolExecutor
from urllib import request
from threading import current_thread
import time
def get(url):
print('%s get %s'%(current_thread().getName(),url))
response=request.urlopen(url)
time.sleep(2)
# print(response.read().decode('utf-8'))
return{'url':url,'content':response.read().decode('utf-8')}
def parse(res):
res=res.result()
print('parse:[%s] res:[%s]'%(res['url'],len(res['content'])))
# get('http://www.baidu.com')
if __name__ == '__main__':
pool=ThreadPoolExecutor(2)
urls=[
'https://www.baidu.com',
'https://www.python.org',
'https://www.openstack.org',
'https://www.openstack.org',
'https://www.openstack.org',
'https://www.openstack.org',
'https://www.openstack.org',
'https://www.openstack.org',
]
for url in urls:
pool.submit(get,url).add_done_callback(parse)
标签：__,Python,self,并发,线程,time,print,进程 From： https://www.cnblogs.com/xuejian123/p/17204374.html