この記事ではPythonの分散プロセスインターフェースを紹介します

[[399138]]

1. はじめに

スレッドとプロセスの間では、プロセスの方が安定しており、複数のマシンに分散できるため、プロセスの方が優先されます。一方、スレッドは、最大で同じマシン上の複数の CPU にしか分散できません。

Python の multiprocessing モジュールは複数のプロセスをサポートするだけでなく、そのマネージャーサブモジュールは複数のプロセスを複数のマシンに分散することもサポートします。サービスプロセスをスケジューラとして記述し、タスクを複数の他のプロセスに分散し、ネットワーク通信を通じて管理することができます。

2. 事例分析

クローラープログラムを実行すると、Web サイトのすべての画像がキャプチャされます。複数のプロセスが使用される場合、通常、1 つのプロセスが画像のリンクアドレスをキャプチャしてキューにリンクアドレスを配置する役割を担い、別のプロセスがキューからリンクアドレスを取得してダウンロードし、ローカルに保存する役割を担います。

分散プロセスでそれを実装するにはどうすればよいでしょうか?

一方のマシン上のプロセスはリンクアドレスを取得する役割を担い、もう一方のマシン上のプロセスはそれを保存する役割を担います。主な問題は、他のマシンプロセスがアクセスできるようにキューをネットワークに公開することです。分散プロセスはこのプロセスをカプセル化します。これは、ローカルキューのネットワーク化と呼ぶことができます。

例：

1.py

 multiprocessing.managersからBaseManager をインポートします
マルチプロセッシングからfreeze_supportをインポートし、キュー
# タスク数
タスク番号 = 10 
 
 # 送信キューと受信キュー
task_queue = キュー(タスク番号)
 result_queue = キュー(タスク番号) 
 
 get_task() を定義します:
 task_queueを返す
 
 get_result() を定義します:
結果キューを返す
# 同様のキューマネージャを作成する
クラス QueueManager(BaseManager):
    合格
 
 win_run() を定義します:
    # ネットワークに登録され、Queue オブジェクトに関連付けられた呼び出し可能オブジェクト
    # キューオブジェクトをネットワーク上に公開する
    # ウィンドウ下のバインディング呼び出しインターフェースはラムダを直接使用できないため、最初に関数を定義してからバインドすることしかできません。
    QueueManager.register( 'get_task_queue' 、呼び出し可能=get_task)
    QueueManager.register( 'get_result_queue' 、呼び出し可能=get_result)
    # ポートをバインドし、認証パスワードを設定する
    マネージャー = QueueManager(アドレス = ( '127.0.0.1' , 8001), authkey = 'qiye' .encode())
    # 管理を開始し、情報チャネルを監視する
    マネージャー.開始() 
 
    試す：
        # ネットワーク経由でタスクキューと結果キューを取得する
        タスク = manager.get_task_queue()
        結果 = manager.get_result_queue() 
 
        # タスクを追加
urlが[ "ImageUrl_" + str(i)の範囲内、 iが範囲 (10)]の範囲内の場合:
            print( 'urlは%sです' % url)
            タスク.put(URL) 
             
        print( '結果を取得してみてください' )
 iが範囲(10)内にある場合:
            print( '結果は%sです' % result.get(timeout=10)) 
 
を除外する：
        print( 'マネージャーエラー' )
    ついに：
        マネージャー.シャットダウン() 
 
 __name__ == '__main__'の場合:
    フリーズサポート()
    win_run()

サーバーに接続し、ポートと認証パスワードがサーバープロセスのものとまったく同じであることを確認します。ネットワークからキューを取得し、ローカライズし、タスクキューからタスクを取得して、結果を結果キューに書き込みます。

2.py

 #コーディング:utf-8
インポート時間 
 multiprocessing.managersからBaseManager をインポートします
# 同様のマネージャーを作成します:
クラス Manager(BaseManager):
    合格
#QueueManagerを使用してQueueのメソッド名を登録および取得する
マネージャー.register( 'get_task_queue' )
 Manager.register( 'get_result_queue' )を登録します。
 #サーバーに接続します:
サーバーアドレス = '127.0.0.1'  
 print( 'サーバー%sに接続...' % server_addr)
 # ポートと認証パスワードがサービス プロセスで設定されたものとまったく同じであることを確認します。
 m = マネージャー(アドレス=(server_addr, 8001), authkey= 'qiye' )
 # ネットワークから接続:
 m.connect () 関数
#キューオブジェクトを取得します:
タスク = m.get_task_queue()
結果 = m.get_result_queue()
 #タスク キューからタスクを取得し、結果を結果キューに書き込みます。
 while(task.empty()ではない):
        image_url = task.get( True 、タイムアウト = 5)
        print( 'タスクを実行 %s をダウンロード...' % image_url)
時間.sleep(1)
        result.put( '%s--->成功' %image_url)
 ＃仕上げる：
 print( 'ワーカー終了。' )