进程池和线程池

先看个例子：

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
image.png