进程池和线程池

先看个例子：

from threading import Thread, current_thread
from queue import Queue
import time


def task_1():
    time.sleep(1)
    print('任务一')

def task_2():
    time.sleep(1)
    print('任务二')

thread = Thread(target=task_1)
thread.start()
thread.join()  #加入join让子进程执行完，主进程再继续往下执行
thread.start()

image.png

线程结束之后就不能再重复使用了。那要执行2个任务，我们可以创建2个线程各执行一个任务，但是我们需要节约资源，那如何只让一个线程执行几个任务呢，即如何重复使用呢？

我们可以利用生产者消费者模型来


def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2():
    print('任务二开始')
    time.sleep(1)
    print('任务二完成')

class MyThread(Thread):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        while True:
            task = self.queue.get()   #任务，子线程作为一个消费者
            print('拿到了：',task)    #获取任务
            task()    #执行任务

q = queue.Queue(5)  #创建队列
thread = MyThread(q)
thread.start()
q.put(task_1)   #主线程充当生产者
q.put(task_2)

运行结果：

image.png

上述代码中，因为队列只是给子进程用，所以可以放到类的初始化方法中，put也可以给子进程中去生产，所以可改写成


def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2():
    print('任务二开始')
    time.sleep(1)
    print('任务二完成')

class MyThread(Thread):
    def __init__(self):
        super().__init__()  
        self.queue = queue.Queue(5)   #创建队列

    def run(self):
        while True:
            task = self.queue.get()   #任务，子线程作为一个消费者
            print('拿到了：',task)    #获取任务
            task()    #执行任务

    def apply_async(self,task):
         self.queue.put(task)  #生产者


thread = MyThread()
thread.start()
thread.apply_async(task_1)   #start调用的只是线程的run方法，所以生产还需要再去调用方法
thread.apply_async(task_2)

运行结果同上。

但是执行完还没结束，一直停留在那，需要开启守护模式。
但是注意设置了守护模式super().__init__(daemon=True) #设置守护模式，主进程结束，会把守护进程杀死运行是没数据的

image.png

这是因为主进程put完之后，没有其他需要执行的了。
所以需要进行阻塞，利用join进程阻塞，使得子进程执行完，主进程再继续执行，在类中重写join方法


def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2():
    print('任务二开始')
    time.sleep(1)
    print('任务二完成')

class MyThread(Thread):
    def __init__(self):
        super().__init__(daemon=True)  #设置守护模式，主进程结束，会把守护进程杀死
        self.queue = queue.Queue(5)   #创建队列

    def run(self):
        while True:
            task = self.queue.get()   #任务，子线程作为一个消费者
            print('拿到了：',task)    #获取任务
            task()    #执行任务
            self.queue.task_done()      #让join计数器-1，表明当前的资源处理完了

    def apply_async(self, task):
         self.queue.put(task)  #生产者


    def join(self):
        self.queue.join()     #等待所有的队列资源都用完，等到队列为空，再执行别的操作

'''
如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束，
在最后执行个join()是等不到结果的，会一直挂起。可以理解为，每task_done一次 就从队列里删掉一个元素，
这样在最后join的时候根据队列长度是否为零来判断队列是否结束，从而执行主线程。
'''

thread = MyThread()
thread.start()
thread.apply_async(task_1)   #start调用的只是线程的run方法，所以生产还需要再去调用方法
thread.apply_async(task_2)
thread.join()    #因为如果设置守护模式，主进程结束，会把守护进程杀死，所以这里调用join等子进程结束之后，主进程再继续执行

运行结果：

image.png

现在可以正常结束了。
传参数形式：


def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2(a,b):
    print('任务二开始')
    print(a,b)
    time.sleep(1)
    print('任务二完成')

class MyThread(Thread):
    def __init__(self):
        super().__init__(daemon=True)  #设置守护模式，主进程结束，会把守护进程杀死
        self.queue = queue.Queue(5)   #创建队列

    def run(self):
        while True:
            task, args, kwargs = self.queue.get()   #任务，子线程作为一个消费者
            print('拿到了：',task)    #获取任务
            task(*args, **kwargs)    #执行任务
            self.queue.task_done()      #让join计数器-1，表明当前的资源处理完了

    def apply_async(self, task, *args, **kwargs):
        self.queue.put((task, args, kwargs))  #生产者

    def join(self):
        self.queue.join()     #等待所有的队列资源都用完，等到队列为空，再执行别的操作

'''
如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束，
在最后执行个join()是等不到结果的，会一直挂起。可以理解为，每task_done一次 就从队列里删掉一个元素，
这样在最后join的时候根据队列长度是否为零来判断队列是否结束，从而执行主线程。
'''

thread = MyThread()
thread.start()
thread.apply_async(task_1)   #start调用的只是线程的run方法，所以生产还需要再去调用方法
thread.apply_async(task_2,2,3)
thread.join()    #因为如果设置守护模式，主进程结束，会把守护进程杀死，所以这里调用join等子进程结束之后，主进程再继续执行

运行结果：

image.png

池

主线程：相当于生产者，只管向线程池提交任务。并不关心线程池是如何执行任务的。因此，并不关心是哪一个线程执行的这个任务。

线程池：相当于消费者，负责接收任务，并将任务分配到一个空闲的线程中去执行。

image.png

池简单实现


def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2(a,b):
    print('任务二开始')
    print(a,b)
    time.sleep(1)
    print('任务二完成')

class MyThread():
    def __init__(self, num):
        self.queue = queue.Queue()   #创建队列
        for i in range(num):
            Thread(target=self.run,daemon=True).start()

    def run(self):
        while True:
            task, args, kwargs = self.queue.get()   #任务，子线程作为一个消费者
            print('拿到了：',task)    #获取任务
            task(*args, **kwargs)    #执行任务
            self.queue.task_done()      #让join计数器-1，表明当前的资源处理完了

    def apply_async(self, task, *args, **kwargs):
        self.queue.put((task, args, kwargs))  #生产者

    def join(self):
        self.queue.join()     #等待所有的队列资源都用完，等到队列为空，再执行别的操作

'''
如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束，
在最后执行个join()是等不到结果的，会一直挂起。可以理解为，每task_done一次 就从队列里删掉一个元素，
这样在最后join的时候根据队列长度是否为零来判断队列是否结束，从而执行主线程。
'''
thread = MyThread(5)
thread.apply_async(task_1)
thread.apply_async(task_2,2,3)
print('任务提交完成')
thread.join()
print('任务完成')

运行结果：

image.png

内置线程池


from multiprocessing.pool import ThreadPool   #线程池

print('-----------内置线程池-----------')

def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2(*args,**kwargs):
    print('任务二开始')
    print(args, kwargs)
    time.sleep(1)
    print('任务二完成')

pool = ThreadPool(2)   #使用内置的pool
pool.apply_async(task_1)   # 向池中提交任务
pool.apply_async(task_2, args=(2,3,4),kwds={'a':5,'b':6})
print('任务提交完成')
pool.close() #在join之前必须close,就不允许再提交任务了。
pool.join()
print('任务完成')

image.png

内置进程池


from multiprocessing import Pool   #进程池

print('-----------内置进程池-----------')

def task_1():
    print('任务一开始')
    time.sleep(1)
    print('任务一完成')

def task_2(*args,**kwargs):
    print('任务二开始')
    print(args, kwargs)
    time.sleep(1)
    print('任务二完成')

pool = Pool(2)   #使用内置的pool
pool.apply_async(task_1)   # 向池中提交任务
pool.apply_async(task_2, args=(2,3,4),kwds={'a':5,'b':6})
print('任务提交完成')
pool.close() #在join之前必须close,就不允许再提交任务了。
pool.join()
print('任务完成')

运行结果：

image.png

使用线程池来实现并发服务器

from multiprocessing.pool import ThreadPool
import socket

server =socket.socket()

server.bind(('0.0.0.0',7001))
server.listen()

print('等待连接......')

def workon(conn):
    while True:
        data = conn.recv(1024)
        if data:
            print('接受数据{}'.format(data.decode()))
            conn.send(data)
        else:
            conn.close()
            break

if __name__ == '__main__':
    pool = ThreadPool(5)
    while True:
        conn, addr = server.accept()
        print('来自{}的连接'.format(addr))
        pool.apply_async(workon,args=(conn,))

使用进程池+线程池来实现并发服务器


from multiprocessing.pool import ThreadPool   #线程池
from multiprocessing import Pool, cpu_count        #进程池
import socket

print('------------使用进程池+线程池来实现并发服务器------------')


server =socket.socket()

server.bind(('0.0.0.0',7001))
server.listen()

print('等待连接......')

def workon_thread(conn):
    while True:
        data = conn.recv(1024)
        if data:
            print('接受数据{}'.format(data.decode()))
            conn.send(data)
        else:
            conn.close()
            break

def workon_process(server):
    thread_pool = ThreadPool(cpu_count()*2)  #通常分配2倍个数的线程
    while True:
        conn, addr = server.accept()
        print('来自{}的连接'.format(addr))
        thread_pool.apply_async(workon_thread,args=(conn,))



if __name__ == '__main__':
    n = cpu_count()        #当前计算机cpu核数
    process_pool = Pool(n)

    for i in range(n):   #充分利用cpu，为每一个cpu分配一个进程
        process_pool.apply_async(workon_process,args=(server,))

process_pool.close()
process_pool.join()

客户端：

import socket

client = socket.socket()

client.connect(('127.0.0.1', 7001))

while True:
    message = input('发送消息>>>')
    if message !='q':
        client.send(message.encode())
        data = client.recv(1024)
        print('接受的消息>>>{}'.format(data.decode()))

    else:
        print('close client socket')
        client.close()
        break

运行结果：

image.png

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 215,133评论 6赞 497
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 91,682评论 3赞 390
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 160,784评论 0赞 350
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 57,508评论 1赞 288
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,603评论 6赞 386
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,607评论 1赞 293
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,604评论 3赞 415
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 38,359评论 0赞 270
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,805评论 1赞 307
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 37,121评论 2赞 330
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 39,280评论 1赞 344
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,959评论 5赞 339
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,588评论 3赞 322
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 31,206评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,442评论 1赞 268
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 47,193评论 2赞 367
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 44,144评论 2赞 352

进程池和线程池

池

池简单实现

内置线程池

内置进程池

使用线程池来实现并发服务器

使用进程池+线程池来实现并发服务器

推荐阅读更多精彩内容