python 之 并发编程(进程池与线程池同步异步阻塞非阻塞线程queue)

Posted mylu

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python 之 并发编程(进程池与线程池同步异步阻塞非阻塞线程queue)相关的知识,希望对你有一定的参考价值。

9.11 进程池与线程池

池子使用来限制并发的任务数目,限制我们的计算机在一个自己可承受的范围内去并发地执行任务

池子内什么时候装进程:并发的任务属于计算密集型 池子内什么时候装线程:并发的任务属于IO密集型

进程池:

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
?
def task(x):
    print(%s 接客 %os.getpid())
    time.sleep(random.randint(2,5))
    return x**2
?
if __name__ == __main__:  # ProcessPoolExecutor创建并开启指定数目的进程
    p=ProcessPoolExecutor() # 默认开启的进程数是cpu的核数
?
    for i in range(20):
        p.submit(task,i)    # 一下并行执行四个任务,等其中一个任务执行完后再执行下一个

线程池:

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
?
def task(x):
    print(%s 接客 %x)
    time.sleep(random.randint(2,5))
    return x**2
?
if __name__ == __main__:  # ThreadPoolExecutor创建并开启指定数目的线程
    p=ThreadPoolExecutor(4) # 默认开启的线程数是cpu的核数*5
?
    for i in range(20):
        p.submit(task,i)    # 一下并发执行四个任务,等其中一个任务执行完后再并发执行下一个

9.112 基于多线程实现并发的套接字通信(使用线程池)

服务端:

技术图片
from socket import *
from threading import Thread
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
?
tpool=ThreadPoolExecutor(3)         #ThreadPoolExecutor创建并开启指定数目的线程
def communicate(conn,client_addr):
    while True:  # 通讯循环
        try:
            data = conn.recv(1024)
            if not data: break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()
?
def server():
    server=socket(AF_INET,SOCK_STREAM)
    server.bind((127.0.0.1,8080))
    server.listen(5)
?
    while True: # 链接循环
        conn,client_addr=server.accept()
        print(client_addr)
        # t=Thread(target=communicate,args=(conn,client_addr))
        # t.start()
        tpool.submit(communicate,conn,client_addr)#一下并发执行3个任务,等其中一个任务执行完后再并发执行下一个
    server.close()
?
if __name__ == __main__:
    server()
View Code

客户端:

技术图片
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect((127.0.0.1,8080))
?
while True:
    msg=input(>>>: ).strip()
    if not msg:continue
    client.send(msg.encode(utf-8))
    data=client.recv(1024)
    print(data.decode(utf-8))
?
client.close()
View Code

9.12 同步异步阻塞非阻塞

阻塞与非阻塞指的是程序的两种运行状态:

阻塞:遇到 I/O 就发生阻塞,程序一旦遇到阻塞操作就会停在原地,并且立刻释放CPU资源

非阻塞(就绪态或运行态):没有遇到 I/O 操作,或者通过某种手段让程序即便是遇到 I/O 操作也不会停在原地,执行其他操作,力求尽可能多的占有CPU

同步与异步指的是提交任务的两种方式:

同步调用:提交完任务后,就在原地等待,直到任务运行完毕后,拿到任务的返回值,才继续执行下一行代码

异步调用:提交完任务后,不在原地等待,直接执行下一行代码

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
#from multiprocessing import Pool
def task(x):
    print(%s 接客 %x)
    time.sleep(random.randint(1,3))
    return x**2
?
if __name__ == __main__:
    # 异步调用
    p=ThreadPoolExecutor(4) # 默认开启的线程数是cpu的核数*5
    obj_l=[]
    for i in range(10):
        obj=p.submit(task,i)
        obj_l.append(obj)
?
    # p.close()
    # p.join()
    p.shutdown(wait=True)# shutdown指的是不能再往进程池内提交任务,wait=True指等待进程池或线程池内所有的任务都运行完毕
    print(obj_l[3].result()) # 9    #最后拿结果
    print()
?
    # 同步调用
    p=ThreadPoolExecutor(4) # 默认开启的线程数是cpu的核数*5
    for i in range(10):
        print(p.submit(task,i).result())
    print()

9.121 异步调用+回调机制

问题:

1、任务的返回值不能得到及时的处理,必须等到所有任务都运行完毕才能统一进行处理

2、解析的过程是串行执行的,如果解析一次需要花费2s,解析9次则需要花费18s

基于进程池:

技术图片
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests
import os
import time
import random
?
def get(url):
    print(%s GET %s %(os.getpid(),url))
    response=requests.get(url)
    time.sleep(random.randint(1,3))
    if response.status_code == 200:
        return response.text
?
def pasrse(obj):            # 干解析的活
    res=obj.result()        # 回调拿结果
    print(%s 解析结果为:%s %(os.getpid(),len(res))) # 4108 解析结果为:2443
?
if __name__ == __main__:
    urls=[
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.python.org,
    ]
?
    pool=ProcessPoolExecutor(4)
    for url in urls:
        obj=pool.submit(get,url)    #parse函数会在obj对应的任务执行完毕后自动执行,会把obj自动传给parse
        obj.add_done_callback(pasrse)   #四个进程并发爬取信息,主进程在执行解析操作
?
    print(主进程,os.getpid())         # 主进程 4108
View Code

基于线程池:

技术图片
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import requests
import os
import time
import random
?
def get(url):
    print(%s GET %s %(current_thread().name,url))
    response=requests.get(url)
    time.sleep(random.randint(1,3))
    if response.status_code == 200:
        return response.text
?
def pasrse(obj):     # 干解析的活
    res=obj.result()
    print(%s 解析结果为:%s %(current_thread().name,len(res)))#ThreadPoolExecutor-0_1 解析结果为:
                                                         #2443
if __name__ == __main__:                              #ThreadPoolExecutor-0_3 解析结果为:2443
    urls=[
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.baidu.com,
        https://www.python.org,
    ]
?
    pool=ThreadPoolExecutor(4)
    for url in urls:
        obj=pool.submit(get,url)    #parse函数会在obj对应的任务执行完毕后自动执行,会把obj自动传给parse
        obj.add_done_callback(pasrse)     #四个线程并发爬取信息,空闲者执行解析操作
    print(主线程,current_thread().name)  #主线程 MainThread
View Code

9.13 线程queue

队列:先进先出 queue.Queue()

import queue
q=queue.Queue(3) 
?
q.put(1)
q.put(2)
q.put(3)
# q.put(4)   阻塞
?
print(q.get())  #1
print(q.get())  #2
print(q.get())  #3

堆栈:后进先出 queue.LifoQueue()

import queue
q=queue.LifoQueue(3) 
?
q.put(a)
q.put(b)
q.put(c)
?
print(q.get())  #c
print(q.get())  #b
print(q.get())  #a

优先级队列:可以以小元组的形式往队列里存值,第一个元素代表优先级,数字越小优先级越高

PriorityQueue()

import queue
q=queue.PriorityQueue(3) 
q.put((10,user1))
q.put((-3,user2))
q.put((-2,user3))
?
print(q.get())  #(-3, ‘user2‘)
print(q.get())  #(-2, ‘user3‘)
print(q.get())  #(10, ‘user1‘)

以上是关于python 之 并发编程(进程池与线程池同步异步阻塞非阻塞线程queue)的主要内容,如果未能解决你的问题,请参考以下文章

Python线程池与进程池

python之进程池与线程池

python并发编程之进程池,线程池concurrent.futures

Python 37 进程池与线程池 协程

python全栈脱产第37天------进程池与线程池协程gevent模块单线程下实现并发的套接字通信

27 Apr 18 GIL 多进程多线程使用场景 线程互斥锁与GIL对比 基于多线程实现并发的套接字通信 进程池与线程池 同步异步阻塞非阻塞