进程池和线程池

1.可以重复利用的线程

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/10/16 18:19
# @Author  : DoubleChina
# @Site    : 
# @File    : PoolTest.py
# @Software: PyCharm
import threading
import queue


class MyThread(threading.Thread):
    def __init__(self):
        super().__init__()
        self.queue = queue.Queue()
        # 触发等待机制,随着主线程一起退出
        self.daemon = True
        # 实例化的时候就直接启动线程
        self.start()

    def run(self):
        # 无限循环，不停执行任务
        while True:
            # 获取任务与参数
            func, args, kwargs = self.queue.get()  # 获取任务,在执行
            # 执行任务
            func(*args, **kwargs)
            # 告诉queue这个任务处理完毕
            self.queue.task_done()

    def apply_async(self, func, args=(), kwargs={}):  # 接受任务
        # 向队列中提交任务
        self.queue.put((func, args, kwargs))

    def join(self, timeout=None):  # 等待所有提交的任务执行完毕
        # 其实是问queue，任务是否执行完毕
        self.queue.join()


def func1():
    print('func1')


def func2():
    print('func2')


t = MyThread()
t.apply_async(func=func1)
t.apply_async(func=func2)
t.join()
# 1.

2.线程池的简单实现

主线程： 相当于生产者，只管向线程池提交任务。并不关心线程池是如何执行任务的。因此，并不关心是哪一个线程执行的这个任务
线程池： 相当于消费者，负责接收任务，并将任务分配到一个空闲的线程中去执行。

自定义一个线程池

import threading
import queue
import time


class Worker(threading.Thread):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue
        # 触发等待机制,随着主线程一起退出
        self.daemon = True
        self.start()

    def run(self):
        # 无限循环，不停执行任务
        while True:
            # 获取任务与参数
            func, args, kwargs = self.queue.get()  # 获取任务,在执行  # 执行任务
            func(*args, **kwargs)
            # 告诉queue这个任务处理完毕
            self.queue.task_done()

    def join(self, timeout=None):  # 等待所有提交的任务执行完毕
        # 其实是问queue，任务是否执行完毕
        self.queue.join()


class MyPool(object):
    def __init__(self, num):
        self.num = num
        self.queue = queue.Queue()
        for i in range(num):
            Worker(self.queue)

    def apply_async(self, func, args=(), kwargs={}):  # 接受任务
        # 向队列中提交任务
        self.queue.put((func, args, kwargs))

    def join(self):
        self.queue.join()


def func1():
    print(threading.current_thread().getName())
    time.sleep(2)


def my_pool_test():
    start_time = time.time()
    t = MyPool(3)  # 实例一个线程池
    for i in range(4):
        t.apply_async(func=func1)
    t.join()
    print('运行时间是{}秒'.format(time.time() - start_time))

3.Python自带的池

# 线程池
from  multiprocessing.pool import ThreadPool
# 进程池
from multiprocessing import Pool
import time
import random

# 可能是一个请求的函数
def worker(msg):
    time.sleep(2)
    # t_start = time.time()
    # data = random.randint(0, 9)
    print(msg)

def test_thread_pool():  # 线程池测试
    msg_list = [1, 2, 3, 4, 5]

    # 实例化一个线程池,2是线程的数量
    pool = ThreadPool(2)

    # map会把系列里面的元素依次传入前面的方法
    # pool.map(worker, msg_list)
    # 放入一个线程
    pool.apply_async(worker, args='1')
    # join之前 关闭线程池，不在提交任务，否则调用join会报错
    pool.close()
    # 等待线程池里面的任务运行完毕
    pool.join()

def test_process_pool():  # 进程池测试
    # 进程池
    process_pool = Pool()
    # process_pool.apply_async(worker, args=(2,))
    msg_list = [1, 2, 3, 4, 5]
    # map_async 放入一个集合,把任务丢给进程池，加了async之后就不会等待运行结束，主进程退出，所有进程结束
    process_pool.map_async(worker, msg_list)
    # 使用map会进行阻塞
    # process_pool.map(worker, msg_list)
    # process_pool.close()
    # process_pool.join()
    print('tttt')

if __name__ == '__main__':
    # test_thread_pool()
    test_process_pool()

4.使用池来实现并发Socket服务器

# 线程池并发编程
from  multiprocessing.pool import ThreadPool
import socket

server = socket.socket()
server.bind(('', 9999))
server.listen(5)


def reable(conn):
    while True:
        recv_data = conn.recv(1024)
        if recv_data:
            print(recv_data)
            conn.send(recv_data)
        else:
            conn.colse()
            break


def socket_thread_pool():
    print('服务端已经启动')
    pool = ThreadPool(2)
    while True:
        conn, addr = server.accept()
        print('客户端{}已经连接'.format(addr))
        pool.apply_async(reable, args=(conn,))