同步/异步异步回调协成线程队列

柔情只为你懂 2021-11-02 00:16 381阅读 0赞

[同步/异步 异步回调 协成 线程队列][_ _]

**目录：**

[**同步/****异步**][Link 1]

**　　　　[异步回调][Link 2]**

**　　　　[协成][Link 3]**

**　　　　****[线程队列][Link 4]**

# 同步|异步: #

线程的三种状态:  
　　1.就绪  
　　2.运行  
　　3.阻塞  
阻塞和非阻塞描述的是运行的状态  
阻塞 :遇到了IO操作,代码卡住,无法执行下一行,CPU会切换到其他任务  
非阻塞 :与阻塞相反,代码正在执行(运行状态) 或处于就绪状态  
  
同步和异步指的是提交任务的方式  
同步 :提交任务必须等待任务完成,才能执行下一行  
异步 :提交任务不需要等待任务完成,立即执行下一行

代码:

![复制代码][copycode.gif]

1 def task():
     2     for i in range(1000000):
     3         i += 1000
     4     print("11111")
     5 
     6 print("start")
     7 task() # 同步提交方式,等函数运行完菜执行下一行
     8 print("end")
     9 
    10 from threading import  Thread
    11 
    12 print("start1")
    13 Thread(target=task).start() # 异步提交,开启线程,然后去执行之后的代码,线程内代码自行执行
    14 print("end1")

![复制代码][copycode.gif]

# 异步回调:任务执行结束后自动调用某个函数 #

![复制代码][copycode.gif]

异步回调:
    　　在发起异步任务后,子进程或子线程完成任务后需要通知任务发起方.通过调用一个函数,all_done_callback(函数名)
    为什么需要回调?
        子进程帮助主进程完成任务,处理任务的结果应该交还给主进程
    其他方式也可以将数据交还给主进程
        1.shutdown  主进程会等到所有任务完成
        2.result函数 会阻塞直到任务完成
        都会阻塞,导致效率降低,所以使用回调
    注意:
        回调函数什么时候被执行? 子进程任务完成时
        谁在执行回调函数? 主进程
    线程的异步回调
        使用方式都相同,唯一的不同是执行回调函数,是子线程在执行(线程间数据共享)

![复制代码][copycode.gif]

### 三种方式: ###

![复制代码][copycode.gif]

1 # 方式1 自己来保存数据 并执行shutdown  仅在多线程
     2 
     3 res = []
     4 def task():
     5     print("%s is 正在打水" % os.getpid())
     6     time.sleep(0.2)
     7     w = "%s 打的水" % os.getpid()
     8     res.append(w)
     9     return w
    10 
    11 if __name__ == '__main__':
    12     for i in range(20):
    13         # 提交任务会返回一个对象  用于回去执行状态和结果
    14         f = pool.submit(task)
    15         print(f.result()) # 方式2   执行result 它是阻塞的直到任务完成  又变成串行了
    16 
    17     print("11111")
    18     # pool.shutdown() # 首先不允许提交新任务 然后等目前所有任务完成后
    19     # print(res)
    20     print("over")
    21 
    22 ====================================================================================
    23 
    24 pool = ThreadPoolExecutor()
    25 
    26 # 方式3 通过回调(什么是回调 任务执行结束后自动调用某个函数)
    27 def task():
    28     print("%s is 正在打水" % os.getpid())
    29     # time.sleep(0.2)
    30     w = "%s 打的水" % os.getpid()
    31     return w
    32 
    33 def task_finish(res):
    34     print("打水完成! %s" % res)
    35 
    36 if __name__ == '__main__':
    37     for i in range(20):
    38         # 提交任务会返回一个对象  用于回去执行状态和结果
    39         f = pool.submit(task)
    40         f.add_done_callback(task_finish) #添加完成后的回调
    41     print("11111")
    42     print("over")

![复制代码][copycode.gif]

### 利用回调完成生产者消费者: ###

![复制代码][copycode.gif]

多进程:

![复制代码][copycode.gif]

1 from  concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
     2 from threading import current_thread
     3 import  os
     4 # 进程池
     5 pool = ProcessPoolExecutor()
     6 # 爬虫:从网络某个地址获取一个HTML文件
     7 import requests # 该模块用于网络(HTTP)请求
     8 
     9 # 生产数据,即生产者
    10 def get_data_task(url):
    11     print(os.getpid(),"正在生产数据!")
    12     # print(current_thread(),"正在生产数据!")
    13 
    14     response = requests.get(url)
    15     text = response.content.decode("utf-8")
    16     print(text)
    17     return text
    18 
    19 # 处理数据,即消费者
    20 def parser_data(f):
    21     print(os.getpid(),"处理数据")
    22     # print(current_thread(), "处理数据")
    23     print("正在解析: 长度%s" % len(f.result()))
    24 
    25 urls = [
    26     "http://www.baidu.com",
    27     "http://www.baidu.com",
    28     "http://www.baidu.com",
    29     "http://www.baidu.com"
    30 ]
    31 
    32 if __name__ == '__main__':
    33     for url in urls:
    34         f = pool.submit(get_data_task,url)
    35         f.add_done_callback(parser_data)  # 回调函数是主进程在执行
    36         # 因为子进程是负责获取数据的,然而数据怎么处理 ,子进程并不知道.应该把数据还给主进程
    37     print("over")

![复制代码][copycode.gif]

多线程:

![复制代码][copycode.gif]

1 from  concurrent.futures import ThreadPoolExecutor
     2 from threading import current_thread
     3 # 进程池
     4 pool = ThreadPoolExecutor()
     5 
     6 # 爬虫:从网络某个地址获取一个HTML文件
     7 import requests # 该模块用于网络(HTTP)请求
     8 
     9 # 生产数据
    10 def get_data_task(url):
    11     # print(os.getpid(),"正在生产数据!")
    12     print(current_thread(),"正在生产数据!")
    13 
    14     response = requests.get(url)
    15     text = response.content.decode("utf-8")
    16     print(text)
    17     return text
    18 
    19 #   处理数据
    20 def parser_data(f):
    21     # print(os.getpid(),"处理数据")
    22     print(current_thread(), "处理数据")
    23     print("正在解析: 长度%s" % len(f.result()))
    24 
    25 urls = [
    26     "http://www.baidu.com",
    27     "http://www.baidu.com",
    28     "http://www.baidu.com",
    29     "http://www.baidu.com"
    30 ]
    31 
    32 if __name__ == '__main__':
    33     for url in urls:
    34         f = pool.submit(get_data_task,url)
    35         f.add_done_callback(parser_data)  # 回调函数是主进程在执行
    36         # 因为子进程是负责获取数据的  然而数据怎么处理 子进程并不知道  应该把数据还给主进程
    37     print("over")

![复制代码][copycode.gif]

# 线程队列: #

普通队列/堆栈队列/优先级队列:

![复制代码][copycode.gif]

import queue
    
    # 普通队列 先进先出
    q = queue.Queue()
    q.put("a")
    q.put("b")
    
    print(q.get())
    print(q.get())
    
    # 堆栈队列  先进后出 后进先出  函数调用就是进栈  函数结束就出栈 递归造成栈溢出
    q2 = queue.LifoQueue()
    q2.put("a")
    q2.put("b")
    print(q2.get())
    
    # 优先级队列
    q3 = queue.PriorityQueue()  # 数值越小优先级越高  优先级相同时 比较大小 小的先取
    q3.put((-100, "c"))
    q3.put((1, "a"))
    q3.put((100, "b"))
    print(q3.get())

![复制代码][copycode.gif]

# 协程:在单线程下由应用程序级别实现并发 #

![复制代码][copycode.gif]

什么是协程？
        协程指的是单线程下由应用程序级别实现的并发
        即把本来由操作系统控制的切换+保存状态，在应用
        程序里实现了
    
        协程的切换vs操作系统的切换
        优点：
            切换速度远快于操作系统
        缺点：
            一个任务阻塞了，其余的任务都无法执行
        ps：只有遇到io才切换到其他任务的协程才能提升
            单线程的执行效率
    
    为何用协程？
        把单个线程的io降到最低，最大限度地提升单个线程的执行效率
    
    如何实现协程？
        from gevent import spawn,monkey;monkey.patch_all()

![复制代码][copycode.gif]

协程的目的是在单线程下实现并发
    为什么出现协程? 因为cpython中,由于GIL而导致同一时间只有一个线程在跑
    意味着:如果你的程序时计算密集,多线程效率也不会提升
          如果是io密集型 没有必要在单线程下实现并发,我会开启多线程来处理io,子线遇到io,cpu切走.
          不能保证一定切到主线
          如果可以,我在遇到io的时候转而去做计算,这样一来可以保证cpu一直在处理你的程序,当然处理时间太长也要切走
    
     总结:单线程下实现并发,是将io阻塞时间用于执行计算,可以提高效率
     原理:一直使用CPU直到超时
    怎么实现单线程并发?
        并发:指的是看起来像是同时运行,实际是在任务间来回切换,同时需要保存执行的状态
        任务一堆代码  可以用函数装起来
        1.如何让两个函数切换执行
            yield可以保存函数的执行状态
            通过生成器可以实现伪并发
            并发不一定提升效率,当任务全是计算时,反而会降低效率
        2.如何知道发生了io, 从而切换执行?
            第三方模块,gevent
    第三方模块 greenlet 可以实现并发 但是不能检测io
    第三方模块 gevent 封装greenlet 可以实现单线程并发,并且能够检测io操作,自动切换

协程的应用场景:
        TCP 多客户端实现方式
        1.来一个客户端就来一个进程     资源消耗较大
        2.来一个客户端就来一个线程     也不能无限开
        3.用进程池 或 线程池  还是一个线程或进程只能维护一个连接
        4.协程 一个线程就可以处理多个客户端  遇到io就切到另一个

![复制代码][copycode.gif]

## 协成实现:单线程实现并发 ##

### 1.yield 把函数做成生成器,生成器会自动保存状态 ###

![复制代码][copycode.gif]

1 # 这是一个进程,默认包含一个主线程
     2 import time
    　　#生成器函数
     3 def task():
     4     while True:
     5         print("task1")
     6         time.sleep(1)#I/O,CPU切走
     7         yield 1
     8 
     9 def task2():
    10     g = task()
    11     while True:
    12         try:
    13             print("task2")
    14             next(g)#next()函数参数传一个可迭代对象
    15         except Exception:
    16             print("任务完成")
    17             break
    18 task2()
    19 打印结果:
    20     task2
    21     task1
    22     task2
    23     task1
    24     task2
    25     task1
    26     ..........

![复制代码][copycode.gif]

### **2.greenlet模块:帮我们封装yield,可以实现任务切换,但是不能检测I/O** ###

![复制代码][copycode.gif]

# 1.实例化greenlet得到一个对象,传入要执行的任务,至少需要两个任务
    # 2.先让某个任务执行起来,使用对象调用switch
    # 3.在任务的执行过程中,手动调用switch来切换

![复制代码][copycode.gif]

1 import greenlet
     2 import time
     3 def task1():
     4     print("task1 1")
     5     time.sleep(2)
     6     g2.switch()
     7     print("task1 2")
     8     g2.switch()
     9 
    10 def task2():
    11     print("task2 1")
    12     g1.switch()
    13     print("task2 2")
    14 
    15 g1 = greenlet.greenlet(task1)
    16 g2 = greenlet.greenlet(task2)
    17 
    18 g1.switch()

![复制代码][copycode.gif]

### **3.gevent:在greenlet的基础上封装检测io操作,自动切换** ###

![复制代码][copycode.gif]

# 1.spawn函数传入你的任务
    # 2.调用join 去开启任务
    # 3.检测io操作需要打monkey补丁,就是一个函数,在程序最开始的地方调用它

![复制代码][copycode.gif]

1 from gevent import monkey
     2 monkey.patch_all()
     3 
     4 import gevent
     5 import time
     6 def eat():
     7     print('eat food 1')
     8     time.sleep(2)
     9     print('eat food 2')
    10 
    11 def play():
    12     print('play 1')
    13     time.sleep(1)
    14     print('play 2')
    15 
    16 g1=gevent.spawn(eat)
    17 g2=gevent.spawn(play)
    18 
    19 gevent.joinall([g1,g2])
    20 print('主')

![复制代码][copycode.gif]

##  协程实现TCP: ##

![复制代码][copycode.gif]

服务端:

![复制代码][copycode.gif]

1 import gevent
     2 from gevent import monkey
     3 monkey.patch_all()
     4 import socket
     5 
     6 server = socket.socket()
     7 # 重用端口
     8 server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
     9 
    10 server.bind(("127.0.0.1",9999))
    11 
    12 server.listen(5)
    13 def data_handler(conn):
    14     print("一个新连接..")
    15     while True:
    16         data = conn.recv(1024)
    17         conn.send(data.upper())
    18 
    19 while True:
    20     conn,addr = server.accept()
    21     # 切到处理数据的任务去执行
    22     gevent.spawn(data_handler,conn)

posted on 2019-06-18 19:33 [女己。][Link 5] 阅读( ...) 评论( ...) [编辑][Link 6] 收藏

转载于:https://www.cnblogs.com/wyf2019/p/11047164.html

[_ _]: https://www.cnblogs.com/wyf2019/p/11047164.html
[Link 1]: https://www.cnblogs.com/xuechengeng/p/9954936.html#tongyi
[Link 2]: https://www.cnblogs.com/xuechengeng/p/9954936.html#yibuhuidiao
[Link 3]: https://www.cnblogs.com/xuechengeng/p/9954936.html#xiecheng
[Link 4]: https://www.cnblogs.com/xuechengeng/p/9954936.html#duilie
[copycode.gif]: /images/20211101/2d9960eaf891478eb6aeb35e286fcc20.png
[Link 5]: https://www.cnblogs.com/wyf2019/
[Link 6]: https://i.cnblogs.com/EditPosts.aspx?postid=11047164