python_进程池以及线程池

Posted 含笑半步颠√

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python_进程池以及线程池相关的知识,希望对你有一定的参考价值。

可以重复利用的线程

直接上代码

from threading import Thread, current_thread
from queue import Queue
# 重写线程类
class MyThread(Thread):
    def __init__(self):
        super().__init__()
        self.daemon = True  # 守护线程
        self.queue = Queue(10)
        self.start()  # 实例化的时候开启线程

    def run(self):  # 子线程只有这一个线程, 从队列里面拿任务
        while True:
            task, args, kwargs = self.queue.get()  # 拿任务 也是元组
            task(*args, **kwargs)  # 可能有,可能没有,所有传入不定长参数
            self.queue.task_done()   # 结束任务

    def apply_async(self, func, args=(), kwargs={}):   # 自写任务,不是重写任务, 充当生产者, 给线程提供任务(把任务扔到队列)
        self.queue.put((func, args, kwargs))

    def join_R(self):   # 主线程等待子线程结束
        self.queue.join()   # task_done 为0 的时候就阻塞


def func():
    print(1, current_thread())

def func2(*args, **kwargs):
    print(2, current_thread())
    print(\'func: \', args, kwargs)

t = MyThread()
t.apply_async(func)
t.apply_async(func2, args=(1,2), kwargs={\'a\':1, \'b\':2})
print("任务提交完成")
t.join_R()
print("任务完成")

结果:

任务提交完成
1 <MyThread(Thread-1, started daemon -1223214272)>
2 <MyThread(Thread-1, started daemon -1223214272)>
func:  (1, 2) {\'a\': 1, \'b\': 2}
任务完成    任务完成后,主线程就开始退出, 因此守护线程被杀死

 

线程池的简单实现

池的概念

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

 

代码实现如下:

from threading import Thread, current_thread
from queue import Queue

class T_pool:
    def __init__(self, n):  # 准备多少个池
        super().__init__()
        self.queue = Queue()
        for i in range(n):  # 在池里开多少个线程
            Thread(target=self.fun, daemon=Thread).start()    # 守护进程 并启动

    def fun(self):   # 生产者
        while True:
            task = self.queue.get()
            task()
            self.queue.task_done()

    def apply_async(self, task):   # 消费者
        self.queue.put(task)

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

def func():
    print(current_thread())

def func2():
    print(current_thread())

p = T_pool(2)
p.apply_async(func)
p.apply_async(func2)
p.join()

结果:

<Thread(Thread-1, started daemon -1223324864)>
<Thread(Thread-1, started daemon -1223324864)>

 

Python自带的池

内置线程池

from multiprocessing.pool import ThreadPool     # 线程池
from multiprocessing import pool  # 进程池
# 内置线程池
def fun(*args, **kwargs):
    print(args, kwargs)

p = ThreadPool(2)   # 直接使用内置的
p.apply_async(fun, args=(1,2), kwds={\'a\':1})
p.close()   # 要求:在join前必须要close,这样就不允许再提交任务了
p.join()

结果:

(1, 2) {\'a\': 1}

 

内置进程池

from multiprocessing import Pool  # 进程池
# 内置进程池
def fun(*args, **kwargs):
    print(args, kwargs)

if __name__ == \'__main__\': # 必须要有一个main测试
    p = Pool(2)   # pool的实例化必须在main测试之下
    p.apply_async(fun, args=(1,2), kwds={\'a\':1})
    p.close()   # 要求:在join前必须要close,这样就不允许再提交任务了
    p.join()

结果:

(1, 2) {\'a\': 1}

 

池的其他操作
操作一: close - 关闭提交通道,不允许再提交任务
操作二: terminate - 中止进程池,中止所有任务
操作三: 结果操作

 

结果操作

from multiprocessing.pool import ThreadPool
import time
def func(n):
    if n == 1:
        return 1
    elif n == 2:
        return 2
    return func(n-1) + func(n-2)

pool = ThreadPool()

a_result = pool.apply_async(func, args=(35,))
print("note1:",time.asctime(time.localtime(time.time())))
result = a_result.get() # 会阻塞,知道结果产生了
print("note2:",time.asctime(time.localtime(time.time())))

结果:

note1: Mon Sep 17 00:07:31 2018
note2: Mon Sep 17 00:07:34 2018

 

使用池来实现并发服务器

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

import socket
from multiprocessing.pool import ThreadPool  # 线程池
from multiprocessing import Pool, cpu_count
\'\'\'
使用线程池来实现
并发服务器
\'\'\'
print(cpu_count())

server = socket.socket()
server.bind((\'0.0.0.0\', 8080))
server.listen(1000)

def work_thread(conn):
    while True:
        data = conn.recv(1000)
        if data:
            print(data)
            conn.send(data)

        else:
            conn.close()
            break

if __name__ == \'__main__\':

    t_pool = ThreadPool(5)  # 使用线程池, 通常分配2倍的cpu个数
    while True:
        conn,addr = server.accept()
        t_pool.apply_async(work_thread, args=(conn,))  # 接收的是个任务, conn做为参数

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

import socket
from multiprocessing.pool import ThreadPool  # 线程池
from multiprocessing import Pool, cpu_count
\'\'\'
使用进程池来实现
并发服务器
\'\'\'
print(cpu_count())

server = socket.socket()
server.bind((\'0.0.0.0\', 9000))
server.listen(1000)

def work_process(server):
    t_pool = ThreadPool(cpu_count()*2)  # 使用线程池, 通常分配2倍的cpu个数
    while True:
        conn,addr = server.accept()
        t_pool.apply_async(work_thread, args=(conn,))  # 接收的是个任务, conn做为参数

def work_thread(conn):
    while True:
        data = conn.recv(1000)
        if data:
            print(data)
            conn.send(data)

        else:
            conn.close()
            break


n = cpu_count()  # 获取当前计算机的CPU核心数量
p = Pool(n)
for i in range(n):  # 充分利用CPU, 为每个CPU分配一个进程
    p.apply_async(work_process, args=(server,))

p.close()
p.join()

 

 

客户端:

import socket

click = socket.socket()
click.connect((\'127.0.0.1\', 8888))

while True:
     data = input("请输入你要发送的数据:")
     click.send(data.encode())
     print("接收到的消息: {}".format(click.recv(1024).decode()))

 

 

总结完毕。

 

作者:含笑半步颠√

博客链接:https://www.cnblogs.com/lixy-88428977

声明:本文为博主学习感悟总结,水平有限,如果不当,欢迎指正。如果您认为还不错,欢迎转载。转载与引用请注明作者及出处。

 

以上是关于python_进程池以及线程池的主要内容,如果未能解决你的问题,请参考以下文章

python进程线程协程以及几种自定义线程池

python---基础知识回顾进程和线程(自定义线程池,上下文管理器和协程的使用)

Python 37 进程池与线程池 协程

Python进程池和线程池

Python进程池

Python3模块concurrent.futures模块,线程池进程池