python-多线程（池）/多进程（池）及线程同步编程

た入场券 2023-02-28 11:14 46阅读 0赞

**目录**

多线程编程

threading模块

直接调用

派生

多线程池

ThreadPoolExecutor

线程同步

多进程编程

Process类

直接调用

派生

多进程池

multiprocessing.pool

ProcessPoolExecutor

全部代码

参考

--------------------

# 多线程编程 #

Python提供了几个用于多线程编程的模块，包括thread、threading 和Queue等。thread 和threading模块允许程序员创建和管理线程。thread 模块提供了基本的线程和锁的支持，而threading提供了更高级别，功能更强的线程管理的功能。Queue模块允许用户创建-一个可以用于多个线程之间共享数据的队列数据结构。

不建议使用thread模块，threading模块更高级，既然选择python语言，自然是不想像c语言，汇编语言去考虑内存管理，寄存器等底层的东西，而是希望调用各种模块快速开发（虽然运行效率不高）。

## threading模块 ##

threading 模块除了包含 \_thread 模块中的所有方法外，还提供其他方法：

*  threading.currentThread(): 返回当前的线程变量。
 *  threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
 *  threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

除了使用方法外，线程模块同样提供了Thread类来处理线程，Thread类提供了以下方法:

*  **run():** 用以表示线程活动的方法。
 *  **start():**启动线程活动。
 *  **join(timeout=None):** 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。
 *  **isAlive():** 返回线程是否活动的。
 *  **getName():** 返回线程名。
 *  **setName(name):** 设置线程名。

### 直接调用 ###

实例化thread类，调用start()、join()函数。

# 直接实例化Thread
    def thread_use():
        nloop = 10
        threads = []
    
        for i in range(0, nloop):
            t = Thread(target=hello, args=('thread' + str(i),))
            threads.append(t)
    
        for t in threads:
            t.start()
    
        for t in threads:
            t.join()

直接使用Thread类，不做参数的变化。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70][] 使用Thread类

### 派生 ###

# 派生Thread
    class myThread(Thread):
        def __init__(self, threadID, name, func, args):
            Thread.__init__(self)
            self.threadID = threadID
            self.name = name
            self.func = func
            self.args = args
    
        def run(self):
            print("开始线程：" + self.name)
            try:
                if self.func:
                    self.func(*self.args)
            finally:
                del self.func, self.args
            print("退出线程：" + self.name)
    
    # 使用派生线程类
    def mythread_use():
        nloop = 10
        threads = []
    
        for i in range(0, nloop):
            t = myThread(i, 'thread' + str(i), hello, args=('thread' + str(i),))
            threads.append(t)
    
        for t in threads:
            t.start()
    
        for t in threads:
            t.join()

主要是修改\_\_init\_\_函数和run函数，对参数和运行方式进行修改。调用方式差不多。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 1][] 派生

## 多线程池 ##

### ThreadPoolExecutor ###

# 使用ThreadPoolExecutor
    def thread_pool_executor_use():
        nloop = 10
        executor = ThreadPoolExecutor(4)
        for i in range(nloop):
            executor.submit(hello, 'thread executor' + str(i))

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 2][] 多线程池

## 线程同步 ##

使用队列进行同步，先进先出。

# 线程同步-生产者，消费者
    q = queue.Queue(maxsize=10)
    
    
    # 生产者
    def Producer(name):
        count = 0
        while True:
            q.put('骨头%s' % count)
            print('%s生产了骨头' % name, count)
            count += 1
            time.sleep(0.5)
            if q.qsize() == q.maxsize:
                exit(0)
    
    
    # 消费者
    def Consumer(name):
        count = 6
        while count:
            count -= 1
            print('%s吃了%s！' % (name, q.get()))
            time.sleep(1)
        print('%s吃饱了!' % name)
        exit(0)
    
    
    # 线程同步
    def thread_synch():
        p = Thread(target=Producer, args=('主人',))
        p.start()
        c = Thread(target=Consumer, args=('二哈',))
        c.start()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 3][] 生产者消费者

# 多进程编程 #

## Process类 ##

和Thread类差不多，3.7有些优化，kill，close函数等。

### 直接调用 ###

# 直接实例化Process
    def process_use():
        nloop = 10
        processes = []
    
        for i in range(0, nloop):
            p = Process(target=hello, args=('process' + str(i),))
            processes.append(p)
    
        for p in processes:
            p.start()
    
        for p in processes:
            p.join()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 4][] Process类

### 派生 ###

同样是修改\_\_init\_\_函数和run函数，不做赘述。

## 多进程池 ##

### multiprocessing.pool ###

常用apply\_async和map函数

# 进程池
    def process_pool_use():
        nloop = 10
        arg_list = []
        with Pool(processes=4) as pool:  # start 4 worker processes
            for i in range(nloop):
                pool.apply_async(hello, ('process'+str(i),))  # evaluate "f(10)" asynchronously in a single process
                arg_list.append('process'+str(i))
            pool.map(hello, arg_list)
            pool.close()
            pool.join()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 5][] 多进程池

### ProcessPoolExecutor ###

# 使用ProcessPoolExecutor
    def process_pool_executor_use():
        nloop = 10
        executor = ProcessPoolExecutor(4)
        for i in range(nloop):
            executor.submit(hello, 'process executor' + str(i))

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 6][] ProcessPoolExecutor

# 全部代码 #

"""
    --coding:utf-8--
    @File: multithreading.py
    @Author:frank yu
    @DateTime: 2020.07.21 18:32
    @Contact: frankyu112058@gmail.com
    """
    from threading import Thread
    from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
    from multiprocessing import Process, Pool
    import queue
    import time
    
    
    # 简单的输出函数
    def hello(con):
        print('hello,' + con)
    
    
    # 派生Thread
    class myThread(Thread):
        def __init__(self, threadID, name, func, args):
            Thread.__init__(self)
            self.threadID = threadID
            self.name = name
            self.func = func
            self.args = args
    
        def run(self):
            print("开始线程：" + self.name)
            try:
                if self.func:
                    self.func(*self.args)
            finally:
                del self.func, self.args
            print("退出线程：" + self.name)
    
    
    # 直接实例化Thread
    def thread_use():
        nloop = 10
        threads = []
    
        for i in range(0, nloop):
            t = Thread(target=hello, args=('thread' + str(i),))
            threads.append(t)
    
        for t in threads:
            t.start()
    
        for t in threads:
            t.join()
    
    
    # 使用派生线程类
    def mythread_use():
        nloop = 10
        threads = []
    
        for i in range(0, nloop):
            t = myThread(i, 'thread' + str(i), hello, args=('thread' + str(i),))
            threads.append(t)
    
        for t in threads:
            t.start()
    
        for t in threads:
            t.join()
    
    
    # 使用ThreadPoolExecutor
    def thread_pool_executor_use():
        nloop = 10
        executor = ThreadPoolExecutor(4)
        for i in range(nloop):
            executor.submit(hello, 'thread executor' + str(i))
    
    
    # 线程同步-生产者，消费者
    q = queue.Queue(maxsize=10)
    
    
    # 生产者
    def Producer(name):
        count = 0
        while True:
            q.put('骨头%s' % count)
            print('%s生产了骨头' % name, count)
            count += 1
            time.sleep(0.5)
            if q.qsize() == q.maxsize:
                exit(0)
    
    
    # 消费者
    def Consumer(name):
        count = 6
        while count:
            count -= 1
            print('%s吃了%s！' % (name, q.get()))
            time.sleep(1)
        print('%s吃饱了!' % name)
        exit(0)
    
    
    # 线程同步
    def thread_synch():
        p = Thread(target=Producer, args=('主人',))
        p.start()
        c = Thread(target=Consumer, args=('二哈',))
        c.start()
    
    
    # 直接实例化Process
    def process_use():
        nloop = 10
        processes = []
    
        for i in range(0, nloop):
            p = Process(target=hello, args=('process' + str(i),))
            processes.append(p)
    
        for p in processes:
            p.start()
    
        for p in processes:
            p.join()
    
    
    # 进程池
    def process_pool_use():
        nloop = 10
        arg_list = []
        with Pool(processes=4) as pool:  # start 4 worker processes
            for i in range(nloop):
                pool.apply_async(hello, ('process' + str(i),))  # evaluate "f(10)" asynchronously in a single process
                arg_list.append('process' + str(i))
            pool.map(hello, arg_list)
            pool.close()
            pool.join()
    
    
    # 使用ProcessPoolExecutor
    def process_pool_executor_use():
        nloop = 10
        executor = ProcessPoolExecutor(4)
        for i in range(nloop):
            executor.submit(hello, 'process executor' + str(i))
    
    
    if __name__ == "__main__":
        # thread_use()
        # mythread_use()
        # thread_pool_executor_use()
        thread_synch()
        # process_use()
        # process_pool_use()
        # process_pool_executor_use()

# 参考 #

python 3.7官方文档:[threading-基于线程的并行][threading-]

python 3.7官方文档:[processing-基于进程的并行][processing-]

python 3.7官方文档：[concurrent.futures--- 启动并行任务][concurrent.futures---]

更多python相关内容：[【python总结】python学习框架梳理][python_python]

本人b站账号:[lady\_killer9][lady_killer9]

有问题请下方评论，转载请注明出处，并附有原文链接，谢谢！如有侵权，请及时联系。如果您感觉有所收获，自愿打赏，可选择支付宝18833895206（小于），您的支持是我不断更新的动力。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70]: /images/20230209/ccddef0fd317403fa5640e26c6d2c191.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 1]: /images/20230209/a5dfc96ef02e4a5fb9b76ada98ed69c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200722155608458.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200722165523117.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/2020072210422352.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200722160323409.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200722113710474.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xhZHlfa2lsbGVyOQ==,size_16,color_FFFFFF,t_70
[threading-]: https://docs.python.org/zh-cn/3.7/library/threading.html
[processing-]: https://docs.python.org/zh-cn/3.7/library/multiprocessing.html
[concurrent.futures---]: https://docs.python.org/zh-cn/3.7/library/concurrent.futures.html
[python_python]: https://blog.csdn.net/lady_killer9/article/details/108763489
[lady_killer9]: https://space.bilibili.com/374432302