python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > python 多进程和多线程

python 多进程和多线程使用详解

作者:酸萝卜别吃

这篇文章主要介绍了python 多进程和多线程使用详解,帮助大家更好得理解和学习使用python,感兴趣的朋友可以了解下

进程和线程

进程是系统进行资源分配的最小单位,线程是系统进行调度执行的最小单位;

一个应用程序至少包含一个进程,一个进程至少包含一个线程;

每个进程在执行过程中拥有独立的内存空间,而一个进程中的线程之间是共享该进程的内存空间的;

Python的多进程

Python的多进程依赖于multiprocess模块;使用多进程可以利用多个CPU进行并行计算;

实例:

from multiprocessing import Process
import os
import time
 
def long_time_task(i):
    print('子进程: {} - 任务{}'.format(os.getpid(), i))
    time.sleep(2)
    print("结果: {}".format(8 ** 20))
 
if __name__=='__main__':
    print('当前母进程: {}'.format(os.getpid()))
    start = time.time()
    p1 = Process(target=long_time_task, args=(1,))
    p2 = Process(target=long_time_task, args=(2,))
    print('等待所有子进程完成。')
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    end = time.time()
    print("总共用时{}秒".format((end - start)))

新创建进程和进程间切换是需要消耗资源的,所以应该控制进程数量;

同时可运行的进程数量收到CPU核数限制;

进程池

使用进程池pool创建进程:

使用进程池可以避免手工进行进程的创建的麻烦,默认数量是CPU核数;

Pool类可以提供指定数量的进程供用户使用,当有新的请求被提交到Pool中的时候,如果进程池还没有满,就会创建一个新的进程来执行请求;如果池已经满了,请求就会等待,等到有空闲进程可以使用时,才会执行请求;

几个方法:

1.apply_async

作用是向进程池提交需要执行的函数和参数,各个进程采用非阻塞的异步方式调用,每个进程只管自己运行,是默认方式;

2.map

会阻塞进程直到返回结果;

3.map_sunc

非阻塞进程;

4.close

关闭进程池,不再接受任务;

5.terminate

结束进程;

6.join

主进程阻塞,直到子进程执行结束;

实例:

from multiprocessing import Pool, cpu_count
import os
import time
 
def long_time_task(i):
    print('子进程: {} - 任务{}'.format(os.getpid(), i))
    time.sleep(2)
    print("结果: {}".format(8 ** 20))
 
if __name__=='__main__':
    print("CPU内核数:{}".format(cpu_count()))
    print('当前母进程: {}'.format(os.getpid()))
    start = time.time()
    p = Pool(4)
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print('等待所有子进程完成。')
    p.close()
    p.join()
    end = time.time()
    print("总共用时{}秒".format((end - start)))

在join之前,必须使用close或者terminate,让进程池不再接受任务;

多进程间的数据通信与共享

通常,进程之间是相互独立的,每个进程都有独立的内存。通过共享内存(nmap模块),进程之间可以共享对象,使多个进程可以访问同一个变量(地址相同,变量名可能不同)。多进程共享资源必然会导致进程间相互竞争,所以应该尽最大可能防止使用共享状态。还有一种方式就是使用队列queue来实现不同进程间的通信或数据共享,这一点和多线程编程类似。

下例这段代码中中创建了2个独立进程,一个负责写(pw), 一个负责读(pr), 实现了共享一个队列queue。

from multiprocessing import Process, Queue
import os, time, random
 
# 写数据进程执行的代码:
def write(q):
    print('Process to write: {}'.format(os.getpid()))
    for value in ['A', 'B', 'C']:
        print('Put %s to queue...' % value)
        q.put(value)
        time.sleep(random.random())
 
# 读数据进程执行的代码:
def read(q):
    print('Process to read:{}'.format(os.getpid()))
    while True:
        value = q.get(True)
        print('Get %s from queue.' % value)
 
if __name__=='__main__':
    # 父进程创建Queue,并传给各个子进程:
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 启动子进程pw,写入:
    pw.start()
    # 启动子进程pr,读取:
    pr.start()
    # 等待pw结束:
    pw.join()
    # pr进程里是死循环,无法等待其结束,只能强行终止:
    pr.terminate()

Python的多线程

python 3中的多进程编程主要依靠threading模块。创建新线程与创建新进程的方法非常类似。threading.Thread方法可以接收两个参数, 第一个是target,一般指向函数名,第二个时args,需要向函数传递的参数。对于创建的新线程,调用start()方法即可让其开始。我们还可以使用current_thread().name打印出当前线程的名字。 

import threading
import time
 
def long_time_task(i):
    print('当前子线程: {} 任务{}'.format(threading.current_thread().name, i))
    time.sleep(2)
    print("结果: {}".format(8 ** 20))
 
if __name__=='__main__':
    start = time.time()
    print('这是主线程:{}'.format(threading.current_thread().name))
    thread_list = []
    for i in range(1, 3):
        t = threading.Thread(target=long_time_task, args=(i, ))
        thread_list.append(t)
    for t in thread_list:
        t.start()
    for t in thread_list:
        t.join()
    end = time.time()
    print("总共用时{}秒".format((end - start)))

多线程间的数据共享

一个进程所含的不同线程间共享内存,这就意味着任何一个变量都可以被任何一个线程修改,因此线程之间共享数据最大的危险在于多个线程同时改一个变量,把内容给改乱了。如果不同线程间有共享的变量,其中一个方法就是在修改前给其上一把锁lock,确保一次只有一个线程能修改它。threading.lock()方法可以轻易实现对一个共享变量的锁定,修改完后release供其它线程使用。

import threading
 
class Account:
    def __init__(self):
        self.balance = 0
 
    def add(self, lock):
        # 获得锁
        lock.acquire()
        for i in range(0, 100000):
            self.balance += 1
        # 释放锁
        lock.release()
 
    def delete(self, lock):
        # 获得锁
        lock.acquire()
        for i in range(0, 100000):
            self.balance -= 1
            # 释放锁
        lock.release()
 
if __name__ == "__main__":
    account = Account()
    lock = threading.Lock()
    # 创建线程
   thread_add = threading.Thread(target=account.add, args=(lock,), name='Add')
    thread_delete = threading.Thread(target=account.delete, args=(lock,), name='Delete')
 
    # 启动线程
   thread_add.start()
    thread_delete.start()
 
    # 等待线程结束
   thread_add.join()
    thread_delete.join()
 
    print('The final balance is: {}'.format(account.balance))

使用queue队列通信-经典的生产者和消费者模型

from queue import Queue
import random, threading, time
 
# 生产者类
class Producer(threading.Thread):
    def __init__(self, name, queue):
        threading.Thread.__init__(self, name=name)
        self.queue = queue
 
    def run(self):
        for i in range(1, 5):
            print("{} is producing {} to the queue!".format(self.getName(), i))
            self.queue.put(i)
            time.sleep(random.randrange(10) / 5)
        print("%s finished!" % self.getName())
 
# 消费者类
class Consumer(threading.Thread):
    def __init__(self, name, queue):
        threading.Thread.__init__(self, name=name)
        self.queue = queue
 
    def run(self):
        for i in range(1, 5):
            val = self.queue.get()
            print("{} is consuming {} in the queue.".format(self.getName(), val))
            time.sleep(random.randrange(10))
        print("%s finished!" % self.getName())
 
def main():
    queue = Queue()
    producer = Producer('Producer', queue)
    consumer = Consumer('Consumer', queue)
 
    producer.start()
    consumer.start()
 
    producer.join()
    consumer.join()
    print('All threads finished!')
 
if __name__ == '__main__':
    main()

对于IO密集型操作,大部分消耗时间其实是等待时间,在等待时间中CPU是不需要工作的,那你在此期间提供双CPU资源也是利用不上的,相反对于CPU密集型代码,2个CPU干活肯定比一个CPU快很多。那么为什么多线程会对IO密集型代码有用呢?这时因为python碰到等待会释放GIL供新的线程使用,实现了线程间的切换。

以上就是python 多进程和多线程使用详解的详细内容,更多关于python 多进程和多线程的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文