[python]マルチスレッドでjsonを返却する方法

Question

やりたいこと
部署IDをマルチスレッドで渡しそれぞれ部署に所属する人の
属性情報を抽出して、最終的にはクライアントにjsonを1つ返却したいです。

{
	"devId" : 111
	personList :[
		{'personId' : a1,
		'attr' : "販売員"},
		{'personId' : a2,
		'attr' : "会計員"},
	] 
},
{
	"devId" : 222
	personList :[
		{'personId' : c1,
		'attr' : "エンジニア"}
	] 
}

dest = dict()
dest.setdefault("items",[])


def lambda_handler(devIdList):

    tpe = ThreadPoolExecutor(max_workers=len(devIdList), thread_name_prefix="thread")
    result = list();

    for devId in devIdList:
        future  = tpe.submit(main,devId)
        result.append(future);
    tpe.shutdown()

    for future in result:
        List.append(future.result())
       
        return {
          'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
          },
          "body": json.dumps([i for i in List])
        }



def main(devId):
    #部署idを渡すと所属員を抽出
    result = GETDevPerson(devId)

    if result is not None:
        if len(result['devIdList']) != 0:
            dest11 = dict()
            dest11.setdefault("devId",devId)
            
             #所属員の詳細情報を抽出
            for personId in result['devIdList']:

                result = GETPersonattribute(personId)
 
                if result is not None:
                    dict1 = {
                             'personId' : personId,
                             'attr' : result['attr']
                    }

 
                    dest11["personList"].append(dict1)
            dest["items"].append(dest11)
            return dest

マルチプロセスを実装しても性能が全然向上しません。
どんなデータが取得できるのか細かく記載していないため
回答が難しいとは思いますが
実装方法に問題があるのでしょうか

Answer 1

極力元のコードから変えないようにするなら、下記のような感じになるでしょうか。
（※18：07に質問が修正される前の質問文に対して作った回答のため、現時点の質問内容に対しては参考にならないかもしれません）

from concurrent.futures import ThreadPoolExecutor, wait
import json

# dest = dict()  削除
# dest.setdefault("items",[])  削除

def lambda_handler(devIdList):

    tpe = ThreadPoolExecutor(max_workers=len(devIdList), thread_name_prefix="thread")
    List = list()

    for devId in devIdList:
        future = tpe.submit(main, devId)
        List.append(future)
    wait(List)
    tpe.shutdown()

    return {
        'headers': {
        'Content-Type': 'application/json',
        'Access-Control-Allow-Origin': '*'
        },
        "body": json.dumps([i.result() for i in List])
    }


def main(devId):
    #部署idを渡すと所属員を抽出
    result = GETDevPerson(devId)
    if result is not None:
        if len(result['devIdList']) != 0:
            dest11 = dict()
            dest11.setdefault("devId", devId)
            dest11.setdefault("personList", [])            

             #所属員の詳細情報を抽出
            for personId in result['devIdList']:

                result = GETPersonattribute(personId)

                if result is not None:
                    dict1 = {
                             'personId' : personId,
                             'attr' : result['attr']
                    }
                    dest11["personList"].append(dict1)
            return dest11

---

並行処理にならない例

from concurrent.futures import ThreadPoolExecutor
import time

def lambda_handler(devIdList):

    tpe = ThreadPoolExecutor(max_workers=len(devIdList))
    List = list()

    for devId in devIdList:
        future = tpe.submit(main, devId)
        # submit直後にfuture.result()を呼び出しているため、futureが完了するまでループはブロックされる（このfutureが完了するまで次のsubmitがなされない）。
        List.append(future.result())

    tpe.shutdown()

    return [i for i in List]
    

def main(devId):
    print(f"main()が呼ばれました. 引数devIDは[{devId}]です。5秒間停止します。")
    time.sleep(5)
    print(f"main()から戻ります。 引数devID：[{devId}]")
    return str(devId)+"_finished." 


idlist = [10, 20, 30]
print(lambda_handler(idlist))

　

並行処理になる例

from concurrent.futures import ThreadPoolExecutor, wait
import time

def lambda_handler(devIdList):

    tpe = ThreadPoolExecutor(max_workers=len(devIdList))
    List = list()

    for devId in devIdList:
        future = tpe.submit(main, devId)
        # futureだけをリストに格納する。result()としていないので、すぐループして次のsubmitが起こる。
        List.append(future)
    wait(List)  # List内のfutureが全部完了するまで待つ（実際はこのwaitの行がなくても、結果自体は変わらない）
    tpe.shutdown()
    # futureの結果をリストにして返す。
    return [i.result() for i in List]
    
import random
def main(devId):
    print(f"main()が呼ばれました. 引数devIDは[{devId}]です。5秒間停止します。")
    time.sleep(5)
    print(f"main()から戻ります。 引数devID：[{devId}]")
    return str(devId)+"_finished." 


idlist = [10,20,30]
print(lambda_handler(idlist))

Comments

[pokemonta]

ありがとうございます
すみません。ソースコードが以前と変わっているのはわかったんですがどのような観点で
変更されているのでしょうか？
私のコードは何がまずいのでしょうか？
マルチスレッドになっていませんでしたか？
（性能が出ないので並列で処理されていないのではと懸念してます）

[pokemonta]

max_workers=の値を2とか10に変えても早くなりません

[退会済みユーザー]

polemontaさんが18：07に質問内容を修正する「前」のコードを見て回答を作ったので、私の回答内容は今となってはあまり参考にならないかもしれません。

修正後のコードは、Listの定義がどこにもなくそのままではエラーになってしまう部分を修正すれば、並行処理はできているようには見えます。

なお、こちらの実行環境としてはローカルでしか試しておりません。
仮にローカルではなくamazon lambdaで実行されている場合は、amazon lambda固有の要因で結果が異なる（性能が出ていない）可能性はあります。（その固有の原因というのはわかりません）

また、明らかにされていないGETDevPersonやGETPersonattributeの処理内容が原因で性能が出ていない可能性もあります。こちらも明らかにされていない以上、原因はわかりません。

[pokemonta]

GETDevPersonやGETPersonattributeでも
並列処理しないと行けないかもしれません　
（打つ手がなくて困ってます。あとは、上記を並列化するぐらいしか）