阻塞
阻塞: 程序运行时,遇到了IO,程序挂起,cpu被切走.
非阻塞
非阻塞: 程序没有遇到IO,程序遇到IO但是我通过某种手段,让cpu强行运行我的程序.
? #同步:
提交一个任务,自任务开始运行直到此任务结束(可能有IO),返回一个返回值之后,我在提交下一个#? #异步:
一次提交多个任务,然后我就直接执行下一行代码.
收取结果 :
1将所有的任务的结果统一回收. 2. 完成一个任务,返回一个结果.
给三个人发布任务:
同步: 先告知第一个人完成写书的任务,我从原地等待,等他两天之后完成了,
告诉完事了,我在发布下一个任务......
异步: 直接将三个任务告知三个人,我就忙我的我,直到三个人完成之后,告知我.
方式一: 异步调用,统一回收结果.
# from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
# import time
# import random
# import os
#
# def task(i):
# print(f'{os.getpid()}开始任务')
# time.sleep(random.randint(1,3))
# print(f'{os.getpid()}任务结束')
# return i
#
# if __name__ == '__main__':
#
# # 异步调用
# pool = ProcessPoolExecutor()
# l1 = []
# for i in range(10):
# obj = pool.submit(task,i)
# l1.append(obj)
#
# pool.shutdown(wait=True)
# print(l1)
# for i in l1:
# print(i.result())
# print('===主')
# 统一回收结果: 我不能马上收到任何一个已经完成的任务的返回值,
我只能等到所有的任务全部结束统一回收.
异步调用
步调用返回值如何接收？
# from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
# import time
# import random
# import os
#
# def task(i):
# print(f'{os.getpid()}开始任务')
# time.sleep(random.randint(1,3))
# print(f'{os.getpid()}任务结束')
# return i
# if __name__ == '__main__':
#
# # 异步调用
# pool = ProcessPoolExecutor()
# for i in range(10):
# pool.submit(task,i)
#
# pool.shutdown(wait=True)
# # shutdown: 让我的主进程等待进程池中所有的子进程都结束任务之后,在执行. 有点类似于join.
# # shutdown: 在上一个进程池没有完成所有的任务之前,不允许添加新的任务.
# # 一个任务是通过一个函数实现的,任务完成了他的返回值就是函数的返回值.
# print('===主'同步调用
# from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
# import time
# import random
# import os
#
# def task(i):
# print(f'{os.getpid()}开始任务')
# time.sleep(random.randint(1,3))
# print(f'{os.getpid()}任务结束')
# return i
# if __name__ == '__main__':
#
# # 同步调用
# pool = ProcessPoolExecutor()
# for i in range(10):
# obj = pool.submit(task,i)
# # obj是一个动态对象,返回的当前的对象的状态,有可能运行中,可能(就绪阻塞),还可能是结束了.
# # obj.result() 必须等到这个任务完成后,返回了结果之后,在执行下一个任务.
# print(f'任务结果:{obj.result()}')
#
# pool.shutdown(wait=True)
# # shutdown: 让我的主进程等待进程池中所有的子进程都结束任务之后,在执行. 有点类似与join.
# # shutdown: 在上一个进程池没有完成所有的任务之前,不允许添加新的任务.
# # 一个任务是通过一个函数实现的,任务完成了他的返回值就是函数的返回值.
# print('===主')异步调用+回调函数
# 版本一:
# from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
# import time
# import random
# import os
# import requests
# def task(url):
# '''模拟的就是爬取多个源代码 一定有IO操作'''
# ret = requests.get(url)
# if ret.status_code == 200:
# return ret.text
# def parse(content):
# '''模拟对数据进行分析 一般没有IO'''
# return len(content)
# if __name__ == '__main__':
# # 开启线程池,并发并行的执行
# url_list = [
# 'http://www.baidu.com',
# 'http://www.JD.com',
# 'http://www.JD.com',
# 'http://www.JD.com',
# 'http://www.taobao.com',
# 'https://www.cnblogs.com/jin-xin/articles/7459977.html',
# 'https://www.luffycity.com/',
# 'https://www.cnblogs.com/jin-xin/articles/9811379.html',
# 'https://www.cnblogs.com/jin-xin/articles/11245654.html',
# 'https://www.sina.com.cn/',
# ]
# pool = ThreadPoolExecutor(4)
# obj_list = []
# for url in url_list:
# obj = pool.submit(task,url)
# obj_list.append(obj)
# pool.shutdown(wait=True)
# for res in obj_list:
# print(parse(res.result()))
# # '''
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# parse(res.result())
# print('===主')
# 版本一:
# 1. 异步发出10个任务,并发的执行,但是统一的接收所有的任务的返回值.(效率低,不能实时的获取结果)
# 2. 分析结果流程是串行,影响效率.
版本二: 针对版本一的缺点2,改进,让串行编程并发或者并行.
# from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
# import time
# import random
# import os
# import requests
#
# def task(url):
# '''模拟的就是爬取多个源代码 一定有IO操作'''
# ret = requests.get(url)
# if ret.status_code == 200:
# return parse(ret.text)
# def parse(content):
# '''模拟对数据进行分析 一般没有IO'''
# return len(content)
# if __name__ == '__main__':
# # 开启线程池,并发并行的执行
# url_list = [
# 'http://www.baidu.com',
# 'http://www.JD.com',
# 'http://www.JD.com',
# 'http://www.JD.com',
# 'http://www.taobao.com',
# 'https://www.cnblogs.com/jin-xin/articles/7459977.html',
# 'https://www.luffycity.com/',
# 'https://www.cnblogs.com/jin-xin/articles/9811379.html',
# 'https://www.cnblogs.com/jin-xin/articles/11245654.html',
# 'https://www.sina.com.cn/',
# ]
# pool = ThreadPoolExecutor(4)
# obj_list = []
# for url in url_list:
# obj = pool.submit(task, url)
# obj_list.append(obj)
# '''
# # 1 在开一个线程进程池,并发并行的处理. 再开一个线程进程池,开销大.
# # 2 将原来的任务扩大,
# 版本一:
# 线程池设置4个线程, 异步发起10个任务,每个任务是通过网页获取源码, 并发执行,
# 最后统一用列表回收10个任务, 串行着分析源码.
# 版本二:
# 线程池设置4个线程, 异步发起10个任务,每个任务是通过网页获取源码+数据分析, 并发执行,
# 最后将所有的结果展示出来.
# 耦合性增强了.
# 并发执行任务,此任务最好是IO阻塞,才能发挥最大的效果
# '''
# pool.shutdown(wait=True)
# for res in obj_list: # [obj1, obj2,obj3....obj10]
# print(res.result())
版本三:
# 基于 异步调用回收所有任务的结果我要做到实时回收结果,
# 并发执行任务每个任务只是处理IO阻塞的,不能增加新得功能.
# 异步调用 + 回调函数
# from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
# import time
# import random
# import os
# import requests
#
# def task(url):
# '''模拟的就是爬取多个源代码 一定有IO操作'''
# ret = requests.get(url)
# if ret.status_code == 200:
# return ret.text
# def parse(obj):
# '''模拟对数据进行分析 一般没有IO'''
# print(len(obj.result()))
#
# if __name__ == '__main__':
# # 开启线程池,并发并行的执行
# url_list = [
# 'http://www.baidu.com',
# 'http://www.JD.com',
# 'http://www.JD.com',
# 'http://www.JD.com',
# 'http://www.taobao.com',
# 'https://www.cnblogs.com/jin-xin/articles/7459977.html',
# 'https://www.luffycity.com/',
# 'https://www.cnblogs.com/jin-xin/articles/9811379.html',
# 'https://www.cnblogs.com/jin-xin/articles/11245654.html',
# 'https://www.sina.com.cn/',
# ]
# pool = ThreadPoolExecutor(4)
# for url in url_list:
# obj = pool.submit(task, url)
# obj.add_done_callback(parse)
#
# '''
# 线程池设置4个线程, 异步发起10个任务,每个任务是通过网页获取源码, 并发执行,
# 当一个任务完成之后,将parse这个分析代码的任务交由剩余的空闲的线程去执行,你这个线程继续去处理其他任务.
# 如果进程池+回调: 回调函数由主进程去执行.
# 如果线程池+回调: 回到函数由空闲的线程去执行.
# 异步 回调是一回事儿?
# 异步站在发布任务的角度,
# 站在接收结果的角度: 回调函数 按顺序接收每个任务的结果,进行下一步处理.
# 异步 + 回调:
# 异步处理的IO类型.
# 回调处理非IO原文地址：https://www.cnblogs.com/SkyRabbit/p/11419278.html
时间： 2024-08-04 15:30:58