异步回调
"""
异步任务使用场景
爬虫
1.从目标站点下载网页数据 本质就是HTML格式字符串
2.用re从字符串中提取出你需要的数据
"""
import requests,re,os
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
# response = requests.get("https://www.baidu.com")
# htm = response.content.decode("utf-8")
# print(re.findall("href=.*?com",htm))
def get_data(url):
print("%s 正在请求%s" % (os.getpid(),url))
response = requests.get(url)
print("%s 请求%s成功" % (os.getpid(),url))
return response
def parser(res):
htm = res.content.decode("utf-8")
ls = re.findall("href=.*?com", htm)
print("解析完成! 共%s个连接" % len(ls))
if __name__ == '__main__':
urls = ["https://www.baidu.com",
"https://www.sina.com",
"https://www.tmall.com",
"https://www.taobao.com",
"https://www.jd.com",
"https://www.python.org",
"https://www.apple.com"]
pool = ProcessPoolExecutor(3)
objs = []
for i in urls:
obj = pool.submit(get_data,i)
# res = obj.result() # 会把任务变成串行
# parser(res)
objs.append(obj)
pool.shutdown() # 请求依然是并发,但是请求的结果不能被立即处理
for i in objs: # 解析数据时串行的
parser(i.result())
pool.shutdown() # 请求依然是并发,但是请求的结果不能被立即处理
# 使用异步回调来处理结果
"""
异步任务使用场景
爬虫
1.从目标站点下载网页数据 本质就是HTML格式字符串
2.用re从字符串中提取出你需要的数据
"""
import requests, re, os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from threading import current_thread
# response = requests.get("https://www.baidu.com")
# htm = response.content.decode("utf-8")
# print(re.findall("href=.*?com",htm))
# def get_data(url):
# print("%s 正在请求%s" % (os.getpid(), url))
# response = requests.get(url)
# print("%s 请求%s成功" % (os.getpid(), url))
# return response
#
#
# def parser(obj):
#
# res = obj.result()
# htm = res.content.decode("utf-8")
# ls = re.findall("href=.*?com", htm)
# print("%s解析完成! 共%s个连接" % (os.getpid(),len(ls)))
#
# if __name__ == '__main__':
# urls = ["https://www.baidu.com",
# "https://www.sina.com",
# "https://www.tmall.com",
# "https://www.taobao.com",
# "https://www.jd.com",
# "https://www.python.org",
# "https://www.apple.com"]
# pool = ProcessPoolExecutor(3)
#
# for i in urls:
# obj = pool.submit(get_data, i)
# # res = obj.result() # 会把任务变成串行
# # parser(res)
# obj.add_done_callback(parser)
"""
什么是回调(函数)
a 交给 b一个任务 b在执行完成后回过头调用了a的一个函数 就称之为回调
为什么需要回调函数?
需要获取异步任务的结果,但是又不应该阻塞(降低效率)
高效的获取任务结果
通常异步任务都会和回调函数一起使用
使用方式:
使用add_done_callback函数()给Future对象绑定一个回调函数
注意:在多进程中回调函数 是交给主进程来执行 而在多线程中 回调函数是谁有空谁执行(不是主线程)
"""
# 线程池中使用异步回调
def get_data(url):
print("%s 正在请求%s" % (current_thread().name, url))
response = requests.get(url)
print("%s 请求%s成功" % (current_thread().name, url))
return response
def parser(obj):
res = obj.result()
htm = res.content.decode("utf-8")
ls = re.findall("href=.*?com", htm)
print("%s解析完成! 共%s个连接" % (current_thread().name,len(ls)))
if __name__ == '__main__':
urls = ["https://www.baidu.com",
"https://www.tmall.com",
"https://www.taobao.com",
"https://www.jd.com",
"https://www.python.org",
"https://www.apple.com"]
pool = ThreadPoolExecutor(3)
for i in urls:
obj = pool.submit(get_data, i)
# res = obj.result() # 会把任务变成串行
# parser(res)
obj.add_done_callback(parser)
线程队列
"""
线程队列
与进程队列的区别 进程队列可以被多进程共享 而线程中的队列 就是一个普通的容器不能进程共享
"""
from queue import Queue,LifoQueue,PriorityQueue
# 1. 先进先出队列
# q = Queue(1)
# q.put("a")
# q.put("b",timeout=1)
#
# print(q.get())
# print(q.get(timeout=2))
# 2.last in first out 后进先出队列(堆栈)
# lq = LifoQueue()
# lq.put("a")
# lq.put("b")
# lq.put("c")
#
#
# print(lq.get())
# print(lq.get())
# print(lq.get())
# 3.优先级队列 取出顺序是 由小到大 优先级可以使数字或字符 只要能够比较大小即可
pq = PriorityQueue()
# pq.put((2,"b"))
# pq.put((3,"c"))
# pq.put((1,"a"))
#
# print(pq.get())
# print(pq.get())
# print(pq.get())
pq.put((["a"],"bdslkfjdsfjd"))
pq.put((["b"],"csdlkjfksdjkfds"))
pq.put((["c"],"asd;kjfksdjfkdsf"))
print(pq.get())
print(pq.get())
print(pq.get())
# print([1] < ["1"])
事件
# 事件
# 是用于协调多个线程工作的,当一个线程要执行某个操作,需要获取另一个线程的状态
# 你要给别人打电话 必须明确知道对方手机买好了
# 作为客户端 要连接服务器 必须明确服务器已经启动了,那么作为启动服务器的一方 如何告知客户端?
# 就通过事件
# """
import time
from threading import Thread
from threading import Event
# 使用变量类完成多线程协作
# is_boot = False
# def start():
# global is_boot
# print("正在启动服务器......")
# time.sleep(5)
# print("服务器启动成功!")
# is_boot = True
#
# def connect():
# while True:
# if is_boot:
# print("连接服务器成功!")
# break
# else:
# print("连接服务器失败!")
# time.sleep(0.5)
#
#
# Thread(target=start).start()
# Thread(target=connect).start()
import time
from threading import Thread
from threading import Event
# 创建一个事件
e = Event() #默认False
def start():
print("正在启动服务器......")
time.sleep(5)
print("服务器启动成功!")
e.set() # 就是把事件的值设置为True
def connect():
# 重试3次
for i in range(3):
print("等待服务器启动....")
e.wait(1) # 会阻塞 直到对方把事件设置为True
if e.isSet():
print("连接成功!")
break
else:
print("连接失败")
else: #如果3次都没成功 就打印这个消息
print("服务器没有启动")
Thread(target=start).start()
Thread(target=connect).start()
单线程下实现并发效果
"""
通过生成器就能完成并发执行
生成器的特点 只要函数中出现了yield该函数就变成了生成器
在执行时与普通函数有什么区别??
"""
def test1():
print(1)
print(2)
print(3)
print(test1())
# 使用生成器 实现单线程并发
import time
def task1():
a = 1
while True:
print("task1 run")
a += 1
print(a)
yield
def task2():
g = task1()
while True:
print("task2 run")
time.sleep(10)
next(g)
task2()
"""
单线程并发 是为了提高效率
对于计算密集型任务 单线程并发 反而降低效率
对于IO密集型 如果可以在执行IO操作的时候 切换到其他计算任务 就能提高CPU占用率 从而提高效率
"""
import time
# def task1():
# a = 0
# for i in range(10000000):
# a += i
# yield
#
# def task2():
# b = 0
# g = task1()
# for i in range(10000000):
# b += i
# next(g)
# s = time.time()
# task2()
# print(time.time()-s)
def task1():
a = 0
for i in range(10000000):
a += i
def task2():
b = 0
for i in range(10000000):
b += i
s = time.time()
task1()
task2()
print(time.time()-s)
协程
"""
协程
可以这么理解是协助线程更高效的工作
本质就是单线程实现并发
也称之为微线程(它比线程更轻量级 单线程下任务的切换 比操作系统切换线程要简单的多)
为什么有 是因为 在CPython中 无法并行执行任务导致效率低 所以我们就需要一种方案 能够将单线程的效率最大化 就是协程
Python中 使用Gevent模块来 实现协程 其能在多个任务间进行切换 而且能够自己检测IO
"""
from gevent import monkey
monkey.patch_all()
import gevent
import time
def task1():
print("task1 run")
time.sleep(10)
print("task1 run")
def task2():
print("task2 run")
print("task2 run")
g1 = gevent.spawn(task1)
g2 = gevent.spawn(task2)
g1.join()
g2.join()
