再び1行のマルチスレッド化

Flaskでのプロジェクトで、サーバーの応答を高速化する必要がありました。 ビューで3つのリモートWebサービスへのリクエストが順番に呼び出されるため、キャッシュからのデータを含むページの読み込み時間は10秒に達しませんでした。 はい、おそらくFlaskは使用する価値のあるフレームワークではありませんが、私たちには持っているものがあります。

それでは始めましょう。 実際のコードを公開することはできないので、アカデミックな例で検討します。



タスク1 3つの関数a、b、cがあり、それらは別々のスレッドで呼び出され、それらの実行結果を待って答えを出す必要があります。

問題1を解決するために、 この翻訳を使用しました。ライブラリの使いやすさに魅了されたからです。

import multiprocessing.dummy as multiprocessing import time def a(): time.sleep(2) return 'a' def b(): time.sleep(2) return 'b' def c(): time.sleep(1) return 'c' p = multiprocessing.Pool() results = p.map(lambda f: f(),[a,b,c]) print(results) p.close() p.join()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コード実行結果：

 ['a', 'b', 'c']
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

すばらしいですが、重大なマイナスがあります。 コードの実行時間は制限されず、すべての手順の結果を待ちます。 問題の定式化を変更します。



タスク2 3つの関数a、b、cがあります。これらの関数は、別々のスレッドで呼び出され、一定の時間後にそれらが完了したかどうかを確認して結果を出します。



解決策として、同じライブラリを使用していますが、すでにmap_async関数を使用しています。 違いは、AsyncResultオブジェクトを返すことです。

 import multiprocessing.dummy as multiprocessing import time def a(): time.sleep(2) return 'a' def b(): time.sleep(2) return 'b' def c(): time.sleep(1) return 'c' p = multiprocessing.Pool() result = p.map_async(lambda f: f(),[a,b,c]) TIMEOUT =3 print(results.get(TIMEOUT)) p.close() p.join()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

TIMEOUT> = 3での実行結果は前の場合と同じですが、少なくとも1つのプロシージャが完了する時間がない場合、TimeoutError例外がスローされます。 しかし、この結果は私には完全には合いませんでした。 実際のところ、私の場合、1つの機能を実行する時間が必要でしたが、問題の時点​​では残りの機能が存在しない可能性があります。



問題3別のスレッドで呼び出す必要がある3つの関数a、b、cがあり、関数aの結果を待ちます。

 import multiprocessing.dummy as multiprocessing import time def a(): time.sleep(2) print(1) return 'a' def b(): time.sleep(3) print(2) return 'b' def c(): time.sleep(1) print(3) return 'c' p = multiprocessing.Pool() results=[] for r in p.imap(lambda f: f(),[a,b,c]): results.append(r) break print(results) p.close() p.join()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

実行結果：

 3 1 ['a'] 2
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ご覧のとおり、3つの機能のうち2つが機能しましたが、優先順位の結果のみが得られました。 2番目の結果を取得するには、imap_unorderedを使用します。

 results=[] for r in p.imap_unordered(lambda f: f(),[a,b,c]): results.append(r) if r =='a': break
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

結果：

 3 1 ['c', 'a'] 2
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

メインスレッドで、最速の1つのスレッドの結果のみが必要な場合はどうなりますか？ 前の例からp.join（）呼び出しを削除し、最初の結果でループを終了するだけで十分です。



今、そのような瞬間があります。 スレッドで機能するmultiprocessing.dummyではなく、プロセスで機能するmultiprocessingモジュールを使用しようとすると、プロセス間通信中に関数をシリアル化できないため、シリアル化エラーcPickle.PicklingErrorが生成されます。 コードを機能させるには、エイリアス関数を入力する必要があります。コードはそれほど美しくありませんが、次のようになります。

 import multiprocessing import time def a(): time.sleep(2) return 'a' def b(): time.sleep(2) return 'b' def c(): time.sleep(1) return 'c' params_mapping = { 'a':a, 'b':b, 'c':c } def func(param): return params_mapping[param]() p = multiprocessing.Pool() results = p.map(func,['a','b','c']) print(results) p.close() p.join()