Python のマルチプロセスの詳細な紹介 (コード例)-Python チュートリアル-php.cn

この記事では、Python のマルチ処理について詳しく紹介 (コード例) しています。参考になる内容です。困っている人は参考にしてください。お役に立てれば幸いです。

このセクションでは、Python のマルチプロセスの学習について説明します。

1. マルチプロセスとマルチスレッドの比較

マルチプロセスマルチプロセシングはマルチスレッドに似ており、両方ともPythonで使用されますParallel しかし、スレッドがあるのに、なぜ Python にマルチプロセッシングがあるのでしょうか? 理由は非常に単純で、スレッドで言及されている GIL.

など、スレッドのいくつかの欠点を補うためです。チュートリアル。

マルチプロセッシングの使用も非常に簡単です。スレッド処理をある程度理解していれば、ここからが楽しい時間です。Python ではマルチプロセッシングとスレッド処理をほぼ同じように使用できるため、簡単に始めることができます。 . プレイも簡単です。コンピューターのマルチコアシステムのパワー!

2. プロセスの追加 Process

import multiprocessing as mp
import threading as td

def job(a,d):
    print('aaaaa')

t1 = td.Thread(target=job,args=(1,2))
p1 = mp.Process(target=job,args=(1,2))
t1.start()
p1.start()
t1.join()
p1.join()

ログイン後にコピー

上記の使用状況の比較コードからわかるように、スレッドとプロセスも同様の方法で使用されます。

Use

使用する場合は main 関数を定義するステートメントを追加する必要があります

if __name__=='__main__':

ログイン後にコピー

完全なアプリケーションコード:

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_test.py
@time: 18/8/26 01:12
"""

import multiprocessing as mp

def job(a, d):
    print a, d

if __name__ == '__main__':
    p1 = mp.Process(target=job, args=(1, 2))
    p1.start()
    p1.join()

ログイン後にコピー

実行環境は次のとおりです。ターミナル環境他の編集ツールでは、実行後に結果が出力されない場合がありますが、ターミナルでの実行後に出力される結果は次のとおりです:

➜  baseLearn python ./process/process_test.py
1 2
➜  baseLearn

ログイン後にコピー

3. ストアドプロセス出力キューの機能

キューは各コアまたはスレッドの操作結果をキューに入れ、各スレッドまたはコアの実行が終了するまで待機し、結果をキューから取り出し、操作のロードを続けます。理由は非常に単純で、複数のスレッドで呼び出された関数は戻り値を持てないため、複数のスレッドの操作の結果を格納するために Queue が使用されます

process_queue.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_queue.py
@time: 18/8/26 01:12
"""

import multiprocessing as mp

# 定义一个被多线程调用的函数，q 就像一个队列，用来保存每次函数运行的结果
def job(q):
    res = 0
    for i in range(1000):
        res += i + i**2 + i**3
    q.put(res)   #queue

if __name__ == '__main__':
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))

    # 分别启动、连接两个线程
    p1.start()
    p2.start()
    p1.join()
    p2.join()

    # 上面是分两批处理的，所以这里分两批输出，将结果分别保存
    res1 = q.get()
    res2 = q.get()

    print res1,res2

ログイン後にコピー

出力結果を出力します:

➜ python ./process/process_queue.py
249833583000 249833583000

ログイン後にコピー

4. プロセスプール

プロセスプールは、実行したいものをプールに入れることを意味し、Python が解決します。複数のプロセス自体の問題。

1. マルチプロセスモジュールのインポート

最初に import multiprocessing を実行し、job()

import multiprocessing as mp

def job(x):
    return x*x

ログイン後にコピー

2. プロセスプールを定義します。 Pool () と map()

次に、Pool

pool = mp.Pool()

ログイン後にコピー

を定義します。プールを取得したら、そのプールを特定の関数に対応させることができます。データを pool にスローすると、プールは関数によって返された値を返します。 Pool と以前の Process の違いは、プールにスローされた 関数には戻り値 があるのに対し、Process の には戻り値 があることです。戻り値ではありません。

次に、map() を使用して結果を取得します。map() には、反復する必要がある関数と値を入力する必要があります。 CPU コアに自動的に割り当てられ、結果が返されます

res = pool.map(job, range(10))

ログイン後にコピー

実行してみましょう

def multicore():
    pool = mp.Pool()
    res = pool.map(job, range(10))
    print(res)
    
if __name__ == '__main__':
    multicore()

ログイン後にコピー

完全なコード:

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_queue.py
@time: 18/8/26 01:12
"""

import multiprocessing as mp

def job(x):
    return x*x  # 注意这里的函数有return返回值

def multicore():
    pool = mp.Pool()
    res = pool.map(job, range(10))
    print(res)
    
if __name__ == '__main__':
    multicore()

ログイン後にコピー

実行結果:

➜  baseLearn python ./process/process_pool.py
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

ログイン後にコピー

3カスタマイズされたコア数

Pool が実際に複数のコアを呼び出しているかどうかをどうやって知ることができるでしょうか?反復回数を増やしてから、CPU 負荷を開いて CPU の動作を確認します。

CPU 負荷を開きます (Mac): [アクティビティモニター] > [CPU] > [CPU 負荷] (1 回クリック)

プールのデフォルトサイズは CPU のコア数です。Pool<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false">def multicore(): pool = mp.Pool(processes=3) # 定义CPU核数量为3 res = pool.map(job, range(10)) print(res)</pre><div class="contentsignin">ログイン後にコピー</div></div>## のprocesses

パラメータを渡すことで、必要なコア数をカスタマイズすることもできます # 4. apply_async()

Poolmap() に加えて、結果を返す方法もあります。それが apply_async()## です。 # .

apply_async()

は値を 1 つだけ渡すことができ、オペレーションには 1 つのコアのみを組み込みますが、値を渡すときは iterable に注意する必要があります。値を渡した後にカンマを追加する必要があり、戻り値を取得するには get() メソッドを使用する必要があります

def multicore():
    pool = mp.Pool() 
    res = pool.map(job, range(10))
    print(res)
    res = pool.apply_async(job, (2,))
    # 用get获得结果
    print(res.get())

ログイン後にコピー

実行結果;

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]  # map()
4 # apply_async()

ログイン後にコピー

概要

Pool
デフォルトの呼び出しは CPU コアの数です。processes パラメータを渡すことで CPU コアの数をカスタマイズできます。
map()
反復パラメーターを入力し、複数の結果を返します
apply_async()
セットのみを入力できますパラメータの数を指定して 1 つの結果を返します。map() の効果を取得したい場合は、

このセクションの内容共有メモリを定義する方法を学びます。

CPU 間の通信を許可できるのは共有メモリのみです。

共有値

Value

を使用して、共有メモリテーブルにデータを保存できます。

import multiprocessing as mp

value1 = mp.Value('i', 0) 
value2 = mp.Value('d', 3.14)

ログイン後にコピー

d および

i パラメーターはデータ型の設定に使用されます。d は倍精度浮動小数点型、i# を表します。 ## は、符号付き 整数型 を表します。 <table> <thead><tr class="firstRow"> <th>Type code</th> <th>C Type</th> <th>Python Type</th> <th>Minimum size in bytes</th> </tr></thead> <tbody> <tr> <td><code>'b' signed char int 1 'B' unsigned char int 1 'u' Py_UNICODE Unicode character 2 'h' signed short int 2 'H' unsigned short int 2 'i' signed int int 2 'I' unsigned int int 2 'l' signed long int 4 'L' unsigned long int 4 'q' signed long long int 8 'Q' unsigned long long int 8 'f' float float 4 'd' double float 8

Shared Array

在Python的 mutiprocessing 中，有还有一个Array类，可以和共享内存交互，来实现在进程之间共享数据。

array = mp.Array('i', [1, 2, 3, 4])

ログイン後にコピー

这里的Array和numpy中的不同，它只能是一维的，不能是多维的。同样和Value 一样，需要定义数据形式，否则会报错。我们会在后一节举例说明这两种的使用方法.

错误形式

array = mp.Array('i', [[1, 2], [3, 4]]) # 2维list

"""
TypeError: an integer is required
"""

ログイン後にコピー

六、进程锁Lock

不加进程锁

让我们看看没有加进程锁时会产生什么样的结果。

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_no_lock.py
@time: 18/8/26 09:22
"""

import multiprocessing as mp
import time

def job(v, num):
    for _ in range(5):
        time.sleep(0.5) # 暂停0.5秒，让输出效果更明显
        v.value += num  # v.value获取共享变量值
        print(v.value)

def multicore():
    v = mp.Value('i', 0)  # 定义共享变量
    p1 = mp.Process(target=job, args=(v, 1))
    p2 = mp.Process(target=job, args=(v, 4)) # 设定不同的number看如何抢夺内存
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == '__main__':
    multicore()

ログイン後にコピー

在上面的代码中，我们定义了一个共享变量v，两个进程都可以对它进行操作。在job()中我们想让v每隔0.1秒输出一次累加num的结果，但是在两个进程p1和p2 中设定了不同的累加值。所以接下来让我们来看下这两个进程是否会出现冲突。

结果打印：

➜  baseLearn python ./process/process_no_lock.py
1
5
9
9
13
13
17
17
18
18
➜  baseLearn

ログイン後にコピー

我们可以看到，进程1和进程2在相互抢着使用共享内存v。

加进程锁

为了解决上述不同进程抢共享资源的问题，我们可以用加进程锁来解决。

首先需要定义一个进程锁

 l = mp.Lock() # 定义一个进程锁

ログイン後にコピー

然后将进程锁的信息传入各个进程中

p1 = mp.Process(target=job, args=(v,1,l)) # 需要将Lock传入
p2 = mp.Process(target=job, args=(v,3,l))

ログイン後にコピー

在job()中设置进程锁的使用，保证运行时一个进程的对锁内内容的独占

def job(v, num, l):
    l.acquire() # 锁住
    for _ in range(5):
        time.sleep(0.1) 
        v.value += num # v.value获取共享内存
        print(v.value)
    l.release() # 释放

ログイン後にコピー

全部代码：

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_lock.py
@time: 18/8/26 09:22
"""

import multiprocessing as mp
import time

def job(v, num, l):
    l.acquire() # 锁住
    for _ in range(5):
        time.sleep(0.5) # 暂停0.5秒，让输出效果更明显
        v.value += num  # v.value获取共享变量值
        print(v.value)
    l.release() # 释放

def multicore():
    l = mp.Lock() # 定义一个进程锁
    v = mp.Value('i', 0)  # 定义共享变量
    p1 = mp.Process(target=job, args=(v, 1, l)) # 需要将lock传入
    p2 = mp.Process(target=job, args=(v, 4, l)) # 设定不同的number看如何抢夺内存
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == '__main__':
    multicore()

ログイン後にコピー

运行一下，让我们看看是否还会出现抢占资源的情况：

结果打印：

➜  baseLearn python ./process/process_lock.py
1
2
3
4
5
9
13
17
21
25

ログイン後にコピー

显然，进程锁保证了进程p1的完整运行，然后才进行了进程p2的运行

Python のマルチプロセスの詳細な紹介 (コード例)

1. マルチプロセスとマルチスレッドの比較

2. プロセスの追加 Process

Use

3. ストアド プロセス出力キューの機能

4. プロセス プール

1. マルチプロセス モジュールのインポート

2. プロセス プールを定義します。 Pool () と map()

3カスタマイズされたコア数

このセクションの内容共有メモリを定義する方法を学びます。

Value

Shared Array

六、进程锁Lock

不加进程锁

加进程锁

3. ストアドプロセス出力キューの機能

4. プロセスプール

1. マルチプロセスモジュールのインポート

2. プロセスプールを定義します。 Pool () と map()