python之进程池和线程池

系统管理员 2022-05-10 14:34 264阅读 0赞

## 进程池和线程池 ##

### 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,))