进程池与线程池

「爱情、让人受尽委屈。」 2022-05-16 01:54 322阅读 0赞

**1.可重复利用的线程**

若存在两个function，常规思路是开两个线程。若使用进程安全队列，将function传入队列，再将队列传入线程，就实现了一个线程处理多个function。

from threading import Thread
    import time
    import queue
    def fun1():
        time.sleep(2)
        print("任务一完成")
    def fun2(*args,**kwargs): #args，kwargs是函数的参数
        time.sleep(4)
        print("任务二完成",args,kwargs)
    class Mythread(Thread):
        def __init__(self,que):
            super().__init__()
            self.que=que #线程安全队列，存放function
            self.daemon=True #设置守护进程
        def run(self):
            while True:
                #线程队列有一个数量计数器，每put计数器+1，但get()不会-1，只有tash_done方法才会-1
                func,args,kwargs=self.que.get() #取得队列中的一个事件
                func(*args,**kwargs) #运行事件
                self.que.task_done() #线程队列计数器-1
        def apply_async(self,item,*args,**kwargs): #将函数名，参数以元组的形式加入队列
            self.que.put( (item,args,kwargs) )
        def join(self):
            self.que.join() #判断计数器是否为0，为0便结束线程
                              # 主函数调用来使队列中的事件全部运行完才继续主进程
    que = queue.Queue()
    t = Mythread(que)
    
    t.apply_async(fun1) #向线程队列中添加任务
    t.apply_async(fun2,1,2,a=3,b=4)
    t.start()
    t.join() #阻塞住，使子进程结束后才运行后续主进程
    print('end')

任务一完成
    任务二完成 (1, 2) {'b': 4, 'a': 3}
    end

**2.线程池的简单实现**

![70][]

主线程将要运行的事件全部添加到线程池中，线程池中的可重复利用的线程可接收这些事件并运行

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

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

from threading import Thread
    import time
    import queue
    
    def fun1():
        time.sleep(2)
        print("任务一完成")
    def fun2(*args,**kwargs):
        time.sleep(2)
        print("任务二完成",args,kwargs)
    
    class Mythreadpool:
        def __init__(self,n): #传递参数，表示线程池中的总线程数
            self.queue=queue.Queue() #生成线程安全队列
            for i in range(n):
                Thread(target=self.worker,daemon=True).start() #直接设置守护进程并开启
        def worker(self):
            while True:
                func,args,kwargs=self.queue.get() #从队列中取事件
                func(*args,**kwargs) #运行事件
                self.queue.task_done() #计数器-1
        def apply_async(self,item,*args,**kwargs): #向队列中添加事件
            self.queue.put(  (item,args,kwargs)  )
        def join(self): #阻塞住，判断计数器是否为零
            self.queue.join()
    
    threadpool = Mythreadpool(4)
    
    threadpool.apply_async(fun1)
    threadpool.apply_async(fun2,1,2,a=3)
    print("任务提交完成")
    threadpool.join()

任务提交完成
    任务一完成
    任务二完成 (1, 2) {'a': 3}

**3.python自带池**

**注意线程池库的导入方式：from multiprocessing.pool import ThreadPool**

**       字典参数的传入方式：kwds=\{\}**

from multiprocessing.pool import ThreadPool
    import  time
    
    def fun1():
        time.sleep(2)
        print("任务一完成")
    def fun2(*args,**kwargs):
        time.sleep(2)
        print("任务二完成",args,kwargs)
    pool = ThreadPool(4)
    pool.apply_async(fun1)
    pool.apply_async(fun2,args=(1,2),kwds={'a':3,'d':4})
    print('提交结束')
    pool.close() #注意，join之前必须close，表示不允许再提交事件
    pool.join()
    print('任务完成')

提交结束
    任务一完成
    任务二完成 (1, 2) {'a': 3, 'd': 4}
    任务完成

类似的，进程中也存在进程池，通过

from multiprocessing import Pool

导入，使用时

pool = Pool(4)

### 池的操作 ###

*  操作一： close - 关闭提交通道，不a再提交任务
 *  操作二： apply\_async – 向池中提交任务
 *  操作三： terminate - 中止进程池，中止所有任务

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

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

服务端

from multiprocessing import Pool #进程池
    from multiprocessing.pool import ThreadPool #线程池
    import socket
    
    sock = socket.socket()
    sock.bind(('',8085))
    sock.listen(100)
    def worker(con):
        while True:
            try:
               data = con.recv(1024)
               if data:
                    print(data)
                    con.send(data)
               else:
                   con.close()
                   break
            except Exception as e:
                print(e)
                con.close()
                break
    
    if __name__ == '__main__':
        pool = ThreadPool(4)
        while True:
                con,addr = sock.accept()
                pool.apply_async(worker,args=(con,))

客户端

import socket
    
    client = socket.socket()
    
    client.connect(('127.0.0.1',8085))
    
    while True:
        data = input('please write the message:')
        client.send(data.encode())
        print(client.recv(1024).decode())
        if data == 'Q' or data == 'q':
            break
    
    client.close()

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

服务端

from multiprocessing import Pool,cpu_count
    from multiprocessing.pool import ThreadPool
    import socket
    
    def worker(con):
        while True:
            try:
               data = con.recv(1024)
               if data:
                    print(data)
                    con.send(data)
               else:
                   con.close()
                   break
            except Exception as e:
                print(e)
                con.close()
                break
    
    def work_process(sock):
        thread_pool=ThreadPool(2*cpu_count()) #一般开核数两倍的线程
        while True:
            con,addr=sock.accept()
            print("已连接",con)
            thread_pool.apply_async(worker,args=(con,))
    if __name__=="__main__":
        sock = socket.socket()
        sock.bind(('', 8086))
        sock.listen(100)
        n= cpu_count() #cpu的核数
        process_pool = Pool(n) #开与核数相等的进程
        for i in range(n):
          process_pool.apply_async(work_process,args=(sock,))
          process_pool.close()
          process_pool.join()

客户端同上

[70]: /images/20220516/80f4bf2e8a5742d9b7084ca358ddf9ef.png