Python多进程编程|极客笔记

Python多进程编程

在这篇文章中，我们将学习如何使用Python实现多进程编程。我们还会讨论其高级概念。

什么是多进程

多进程是系统同时运行一个或多个进程的能力。简单来说，多进程使用计算机系统中的两个或更多个CPU。该方法还能够在多个进程之间分配任务。

处理单元共享主存储器和外围设备以同时处理程序。多进程应用程序会被分为更小的部分并独立运行。操作系统通过将每个进程分配给处理器来完成此任务。

Python提供了名为multiprocessing的内置包，支持进程切换。在使用多进程编程之前，我们必须了解进程对象。

为什么使用多进程

多进程对于在计算机系统内执行多个任务至关重要。假设一台计算机没有多进程或只有一个处理器。我们同时为该系统分配各种进程。

系统必须中断之前的任务并转移到另一个任务，以保持所有任务进行。就像一个厨师独自在厨房工作一样简单。他需要做很多任务，如切菜、清洗、烹饪、揉面团、烘烤等。

因此，多进程是在不中断的情况下同时执行多个任务的关键。它还简化了对所有任务的跟踪。这就是为什么出现多进程概念的原因。

多进程可以被表示为具有多个中央处理器的计算机。
多核处理器是指具有两个或更多独立单元的单个计算组件。

在多进程中，每个任务都有自己的处理器，因此CPU可以分配多个任务。

Python中的多进程

Python提供了multiprocessing模块，以在单个系统中执行多个任务。它提供了一个用户友好和直观的API来处理多进程。

让我们来理解一下多进程的简单示例。

示例 –

from multiprocessing import Process
   def disp():
      print ('Hello !! Welcome to Python Tutorial')
      if __name__ == '__main__':
      p = Process(target=disp)
      p.start()
      p.join()

输出：

'Hello !! Welcome to Python Tutorial'

说明：

在上面的代码中，我们导入了Process类，然后在disp()函数内创建了Process对象。然后我们使用start()方法启动进程，并使用join()方法完成进程。我们还可以使用args关键字在声明的函数中传递参数。

下面我们来理解一个带参数的多进程的示例。

示例2

# Python multiprocessing example
# importing the multiprocessing module

import multiprocessing
def cube(n):
   # This function will print the cube of the given number
   print("The Cube is: {}".format(n * n * n))

def square(n):
    # This function will print the square of the given number
   print("The Square is: {}".format(n * n))

if __name__ == "__main__":
   # creating two processes
   process1 = multiprocessing.Process(target= square, args=(5, ))
   process2 = multiprocessing.Process(target= cube, args=(5, ))

   # Here we start the process 1
   process1.start()
   # Here we start process 2
   process2.start()

   # The join() method is used to wait for process 1 to complete
   process1.join()
   # It is used to wait for process 1 to complete
   process2.join()

   # Print if both processes are completed
   print("Both processes are finished")

输出：

The Cube is: 125
The Square is: 25
Both processes are finished

说明 –

在上面的示例中，我们创建了两个函数 – cube() 函数计算给定数字的立方，和 square() 函数计算给定数字的平方。

接下来，我们定义了一个 Process 类的 process 对象，它有两个参数。第一个参数是一个 target，表示要执行的函数，第二个参数是一个 args，表示要在函数中传递的参数。

process1 = multiprocessing.Process(target= square, args=(5, ))
process2 = multiprocessing.Process(target= cube, args=(5, ))

我们使用 start() 方法来启动进程。

process1.start()
process2.start()

正如我们在输出中看到的，它等待 过程一 完成，然后 过程二 。最后一条语句在两个过程都完成后执行。

Python多进程类

Python的multiprocessing模块提供了许多常用于构建并行程序的类。我们将讨论它的主要类 – Process、Queue和Lock。我们已经在前面的示例中讨论了Process类。现在我们将讨论Queue和Lock类。

让我们来看一个简单的示例，获取当前系统中的CPU数量。

示例

import multiprocessing
print("The number of CPU currently working in system : ", multiprocessing.cpu_count())

输出：

('The number of CPU currently woking in system : ', 32)

上述的CPU数量可能因您的计算机而异。对我们来说，核心数是32。

使用Queue类进行Python多进程

我们知道队列是数据结构的重要部分。Python多进程与数据结构队列完全相同，基于”先进先出”的概念。队列通常存储Python对象，在进程之间共享数据起着重要的作用。

队列作为参数传递给进程的目标函数，允许进程消费数据。队列提供了put()函数来插入数据和get()函数来从队列中获取数据。让我们理解以下示例。

示例

# Importing Queue Class

from multiprocessing import Queue

fruits = ['Apple', 'Orange', 'Guava', 'Papaya', 'Banana']
count = 1
# creating a queue object
queue = Queue()
print('pushing items to the queue:')
for fr in fruits:
    print('item no: ', count, ' ', fr)
    queue.put(fr)
    count += 1

print('\npopping items from the queue:')
count = 0
while not queue.empty():
    print('item no: ', count, ' ', queue.get())
    count += 1

输出：

pushing items to the queue:
('item no: ', 1, ' ', 'Apple')
('item no: ', 2, ' ', 'Orange')
('item no: ', 3, ' ', 'Guava')
('item no: ', 4, ' ', 'Papaya')
('item no: ', 5, ' ', 'Banana')

popping items from the queue:
('item no: ', 0, ' ', 'Apple')
('item no: ', 1, ' ', 'Orange')
('item no: ', 2, ' ', 'Guava')
('item no: ', 3, ' ', 'Papaya')
('item no: ', 4, ' ', 'Banana')

解释 –

在上面的代码中，我们导入了Queue类并初始化了名为fruits的列表。接下来，我们将计数（count）赋值为1。计数变量将计算元素的总数。然后，我们通过调用Queue()方法创建了队列对象。该对象将用于在队列中进行操作。在for循环中，我们使用put()函数逐个将元素插入队列，并在每次循环迭代时将计数增加1。

Python Multiprocessing Lock类

multiprocessing Lock类用于在进程上获取锁，以便我们可以阻止其他进程执行相似的代码，直到锁被释放。Lock类主要执行两个任务：第一个是使用acquire()函数获取锁，第二个是使用release()函数释放锁。

Python Multiprocessing示例

假设我们有多个任务。因此，我们创建两个队列：第一个队列将维护任务，另一个队列将存储完成的任务日志。下一步是实例化用于完成任务的进程。如前所述，Queue类已经同步，因此我们不需要使用Lock类获取锁。

在下面的示例中，我们将所有的多进程类合并在一起。让我们看下面的示例。

示例

from multiprocessing import Lock, Process, Queue, current_process
import time
import queue 


def jobTodo(tasks_to_perform, complete_tasks):
    while True:
        try:

            # The try block to catch task from the queue.
            # The get_nowait() function is used to
            # raise queue.Empty exception if the queue is empty.

            task = tasks_to_perform.get_nowait()

        except queue.Empty:

            break
        else:

                # if no exception has been raised, the else block will execute
                # add the task completion


            print(task)
            complete_tasks.put(task + ' is done by ' + current_process().name)
            time.sleep(.5)
    return True


def main():
    total_task = 8
    total_number_of_processes = 3
    tasks_to_perform = Queue()
    complete_tasks = Queue()
    number_of_processes = []

    for i in range(total_task):
        tasks_to_perform.put("Task no " + str(i))

    # defining number of processes
    for w in range(total_number_of_processes):
        p = Process(target=jobTodo, args=(tasks_to_perform, complete_tasks))
        number_of_processes.append(p)
        p.start()

    # completing process
    for p in number_of_processes:
        p.join()

    # print the output
    while not complete_tasks.empty():
        print(complete_tasks.get())

    return True


if __name__ == '__main__':
    main()

输出:

Task no 2
Task no 5
Task no 0
Task no 3
Task no 6
Task no 1
Task no 4
Task no 7
Task no 0 is done by Process-1
Task no 1 is done by Process-3
Task no 2 is done by Process-2
Task no 3 is done by Process-1
Task no 4 is done by Process-3
Task no 5 is done by Process-2
Task no 6 is done by Process-1
Task no 7 is done by Process-3

Python多进程池

Python的多进程池在多个输入值上并行执行函数是必不可少的。它也用于在进程之间分发输入数据 （数据并行） 。考虑以下多进程池的示例。

示例

from multiprocessing import Pool
import time

w = (["V", 5], ["X", 2], ["Y", 1], ["Z", 3])


def work_log(data_for_work):
    print(" Process name is %s waiting time is %s seconds" % (data_for_work[0], data_for_work[1]))
    time.sleep(int(data_for_work[1]))
    print(" Process %s Executed." % data_for_work[0])


def handler():
    p = Pool(2)
    p.map(work_log, w)

if __name__ == '__main__':
    handler()

输出：

Process name is V waiting time is 5 seconds
Process V Executed.
Process name is X waiting time is 2 seconds
Process X Executed.
Process name is Y waiting time is 1 seconds
Process Y Executed.
Process name is Z waiting time is 3 seconds
Process Z Executed.

让我们来了解多进程池的另一个示例。

示例 2

from multiprocessing import Pool
def fun(x):
    return x*x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(fun, [1, 2, 3]))

输出：

[1, 8, 27]

代理对象

代理对象是指驻留在不同进程中的共享对象。该对象也被称为代理。多个代理对象可以具有相同的引用。代理对象包含了各种方法，用于调用其引用对象的相应方法。下面是代理对象的示例。

示例

from multiprocessing import Manager
manager = Manager()
l = manager.list([i*i for i in range(10)])
print(l)
print(repr(l))
print(l[4])
print(l[2:5])

输出：

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
<ListProxy object, typeid 'list' at 0x7f063621ea10>
16
[4, 9, 16]

代理对象是可序列化的，所以我们可以在进程之间传递它们。这些对象还用于控制同步的级别。

常用的多进程函数

到目前为止，我们已经讨论了使用Python进行多进程的基本概念。多进程本身就是一个广泛的主题，对于在单个系统内执行各种任务至关重要。我们定义了一些常用的函数，用于实现多进程。

方法	描述
pipe()	pipe()函数返回一对连接对象。
run()	run()方法用于表示进程活动。
start()	start()方法用于启动进程。
join([timeout])	join()方法用于阻塞进程，直到调用它的进程终止。timeout是可选参数。
is_alive()	返回进程是否存活。
terminate()	如其名，用于终止进程。请记住-在Linux中使用terminate()方法，在Windows中使用TerminateProcess()方法。
kill()	此方法与terminate()类似，但在Unix上使用SIGKILL信号。
close()	此方法用于关闭Process对象并释放与其关联的所有资源。
qsize()	返回队列的近似大小。
empty()	如果队列为空，返回True。
full()	如果队列已满，则返回True。
get_await()	此方法等效于get(False)。
get()	该方法用于从队列获取元素。它从队列中移除并返回一个元素。
put()	该方法用于将一个元素插入队列。
cpu_count()	返回系统中工作CPU的数量。
current_process()	返回与当前进程对应的Process对象。
parent_process()	返回与当前进程对应的父进程Process对象。
task_done()	此函数用于指示已排队的任务已完成。
join_thread()	该方法用于加入后台线程。