프로세스

Python의 다중 스레드는 실제로 실제 다중 스레드가 아닙니다. 멀티 코어 CPU의 리소스를 최대한 활용하려면 Python에서 대부분의 경우 다중 프로세스를 사용해야 합니다. Python은 매우 사용하기 쉬운 다중 처리 패키지를 제공합니다. 함수만 정의하면 Python이 나머지 모든 작업을 수행합니다. 이 패키지를 사용하면 단일 프로세스에서 동시 실행으로의 전환을 쉽게 수행할 수 있습니다. 멀티프로세싱은 하위 프로세스, 통신 및 데이터 공유를 지원하고 다양한 형태의 동기화를 수행하며 프로세스, 큐, 파이프 및 잠금과 같은 구성 요소를 제공합니다.

1. 프로세스 클래스

프로세스를 생성하는 클래스: Process([group [, target [, name [, args [, kwargs]]]])

target은 호출 개체를 나타냅니다.

args는 나타냅니다. 호출 객체 위치 매개변수 튜플

kwargs는 호출 객체의 사전을 나타냅니다.

name은 별칭입니다.

group은 기본적으로 사용되지 않습니다.

함수를 생성하고 이를 여러 프로세스로 사용하는 예를 살펴보겠습니다.

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import multiprocessing
import time
def worker(interval, name):
    print(name + '【start】')
    time.sleep(interval)
    print(name + '【end】')
if __name__ == "__main__":
    p1 = multiprocessing.Process(target=worker, args=(2, '两点水1'))
    p2 = multiprocessing.Process(target=worker, args=(3, '两点水2'))
    p3 = multiprocessing.Process(target=worker, args=(4, '两点水3'))
    p1.start()
    p2.start()
    p3.start()
    print("The number of CPU is:" + str(multiprocessing.cpu_count()))
    for p in multiprocessing.active_children():
        print("child   p.name:" + p.name + "\tp.id" + str(p.pid))
    print("END!!!!!!!!!!!!!!!!!")

출력 결과:

다중 프로세스 출력 결과

2. 프로세스를 클래스로 생성

물론 프로세스 p 호출이 시작되면 아래 예와 같이 클래스로 프로세스를 생성할 수도 있습니다( ), run() 메서드가 자동으로 호출됩니다.

# -*- coding: UTF-8 -*-
import multiprocessing
import time
class ClockProcess(multiprocessing.Process):
    def __init__(self, interval):
        multiprocessing.Process.__init__(self)
        self.interval = interval
    def run(self):
        n = 5
        while n > 0:
            print("当前时间: {0}".format(time.ctime()))
            time.sleep(self.interval)
            n -= 1
if __name__ == '__main__':
    p = ClockProcess(3)
    p.start()

출력 결과는 다음과 같습니다.

Create process class

3.Daemon attribute

daemon 속성이 무엇에 사용되는지 알고 싶다면 다음 두 가지 예를 살펴보세요. daemon 속성이 추가된 것과 없는 경우 출력 비교 결과:

deamon 속성을 추가하지 않은 예:

# -*- coding: UTF-8 -*-
import multiprocessing
import time
def worker(interval):
    print('工作开始时间：{0}'.format(time.ctime()))
    time.sleep(interval)
    print('工作结果时间：{0}'.format(time.ctime()))
if __name__ == '__main__':
    p = multiprocessing.Process(target=worker, args=(3,))
    p.start()
    print('【EMD】')

출력 결과:

【EMD】
工作开始时间：Mon Oct  9 17:47:06 2017
工作结果时间：Mon Oct  9 17:47:09 2017

위 예에서 프로세스 p는 데몬 속성을 추가합니다:

# -*- coding: UTF-8 -*-
import multiprocessing
import time
def worker(interval):
    print('工作开始时间：{0}'.format(time.ctime()))
    time.sleep(interval)
    print('工作结果时间：{0}'.format(time.ctime()))
if __name__ == '__main__':
    p = multiprocessing.Process(target=worker, args=(3,))
    p.daemon = True
    p.start()
    print('【EMD】')

출력 결과:

【EMD】

It 출력 결과를 보면 알 수 있는데, 자식 프로세스에 daemon 속성을 추가하면 메인 프로세스가 종료되면 자식 프로세스도 종료됩니다. 따라서 하위 프로세스에 대한 정보는 인쇄되지 않습니다.

4. 조인 방법

위의 예를 계속해서 자식 스레드의 실행을 끝내려면 어떻게 해야 할까요?

그런 다음 조인 메서드를 사용할 수 있습니다. 조인 메서드의 주요 기능은 조인 메서드를 호출하는 프로세스가 실행을 마칠 때까지 현재 프로세스를 차단한 다음 현재 프로세스를 계속 실행하는 것입니다.

그러면 조인 메소드를 추가하는 예를 보세요:

import multiprocessing
import time
def worker(interval):
    print('工作开始时间：{0}'.format(time.ctime()))
    time.sleep(interval)
    print('工作结果时间：{0}'.format(time.ctime()))
if __name__ == '__main__':
    p = multiprocessing.Process(target=worker, args=(3,))
    p.daemon = True
    p.start()
    p.join()
    print('【EMD】')

출력 결과:

工作开始时间：Tue Oct 10 11:30:08 2017
工作结果时间：Tue Oct 10 11:30:11 2017
【EMD】

5, Pool

자식 프로세스가 많이 필요한 경우 하나씩 생성해야 합니까?

물론 그렇지 않습니다. 프로세스 풀 방법을 사용하여 일괄적으로 하위 프로세스를 생성할 수 있습니다.

예제는 다음과 같습니다.

# -*- coding: UTF-8 -*-
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
    print('进程的名称：{0} ；进程的PID: {1} '.format(name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('进程 {0} 运行了 {1} 秒'.format(name, (end - start)))
if __name__ == '__main__':
    print('主进程的 PID：{0}'.format(os.getpid()))
    p = Pool(4)
    for i in range(6):
        p.apply_async(long_time_task, args=(i,))
    p.close()
    # 等待所有子进程结束后在关闭主进程
    p.join()
    print('【End】')

출력 결과는 다음과 같습니다.

主进程的 PID：7256
进程的名称：0 ；进程的PID: 1492
进程的名称：1 ；进程的PID: 12232
进程的名称：2 ；进程的PID: 4332
进程的名称：3 ；进程的PID: 11604
进程 2 运行了 0.6500370502471924 秒
进程的名称：4 ；进程的PID: 4332
进程 1 运行了 1.0830621719360352 秒
进程的名称：5 ；进程的PID: 12232
进程 5 运行了 0.029001712799072266 秒
进程 4 运行了 0.9720554351806641 秒
进程 0 运行了 2.3181326389312744 秒
进程 3 运行了 2.5331451892852783 秒
【End】

여기서 주의할 점이 한 가지 있습니다. Pool 개체의 Join() 메서드를 호출하면 모든 하위 프로세스가 실행이 완료될 때까지 기다립니다. 그리고 Join()을 호출하기 전에 close()를 호출해야 합니다. close()를 호출한 후에는 계속해서 새 프로세스를 추가할 수 없습니다.

출력 결과에 주의하세요. 하위 프로세스 0, 1, 2, 3은 즉시 실행되는 반면, 하위 프로세스 4는 실행되기 전에 이전 하위 프로세스가 완료될 때까지 기다려야 합니다. 이는 Pool의 기본 크기 때문입니다. 내 컴퓨터에는 4개가 있으므로 최대 4개의 프로세스를 동시에 실행할 수 있습니다. 이는 운영 체제의 제한이 아닌 풀의 의도적인 설계 제한입니다.

p = Pool(5)

로 변경하면 동시에 5개의 프로세스를 실행할 수 있습니다.

6. 프로세스 간 통신

프로세스는 확실히 통신해야 합니다. 운영 체제는 프로세스 간 통신을 달성하기 위해 많은 메커니즘을 제공합니다. Python의 다중 처리 모듈은 기본 메커니즘을 래핑하고 데이터 교환을 위한 큐 및 파이프와 같은 여러 메서드를 제공합니다.

대기열을 예로 들어 상위 프로세스에 두 개의 하위 프로세스를 만듭니다. 하나는 대기열에 데이터를 쓰고 다른 하나는 대기열에서 데이터를 읽습니다.

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
from multiprocessing import Process, Queue
import os, time, random
def write(q):
    # 写数据进程
    print('写进程的PID:{0}'.format(os.getpid()))
    for value in ['两点水', '三点水', '四点水']:
        print('写进 Queue 的值为：{0}'.format(value))
        q.put(value)
        time.sleep(random.random())
def read(q):
    # 读取数据进程
    print('读进程的PID:{0}'.format(os.getpid()))
    while True:
        value = q.get(True)
        print('从 Queue 读取的值为：{0}'.format(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()

출력 결과는 다음과 같습니다.

读进程的PID:13208
写进程的PID:10864
写进 Queue 的值为：两点水
从 Queue 读取的值为：两点水
写进 Queue 的值为：三点水
从 Queue 读取的值为：三点水
写进 Queue 的值为：四点水
从 Queue 读取的值为：四点水