pythonでマルチスレッドの動作のさせ方

Question

pythonでマルチスレッドでSQLを25個実行したいです。
スレッド数は10個まで、という制約があります。最初にSQLを10個なげ、処理が終わったスレッドから次のSQLを投げて行く、というものを作成したいと思っていますが、製造方法がよくわかりません。。

以下のようなイメージで作成したいと思っています。

def excu_sql(n):
    sql_list = ['sql1', 'sql2','sql3', ~]
    resutl = sql_excu(sql_list[n])  # sql実行
    print(result)

def multi_thread():
    for i in range(10):
        print("スレッド" + str(i) + "を開始します")

        excu_sql(n)  # ここで１０個のスレッドがnの値を２５個になるまで順次実施する

必ずしも上記のような形でなくても良いのですが、調べても分からなかったため、投稿させて頂きました。
ちなみに以下のサイトを参考に作成してみたところ、マルチスレッドにならず、シングルスレッド(forと同じ挙動)になりました。。
https://qiita.com/YutaroYamanaka/items/25390222fb07376f515a

Comments

[toast-uz]

マルチスレッドで処理したいのはDBのI/O待ちネックを解消して、全体の処理速度を改善したいということでしょうか？
10スレッド限定はDB事情ではなくクライアント事情でしょうか？DB事情であってもコネクション数ネックなのか、同時クエリー数のネックなのか分かりますでしょうか？
質問者様は並列処理と非同期処理の区別はついていますでしょうか？
事象的にはマルチスレッドによる並列処理ではなく、I/Oを非同期処理にする方が、よい手法であると思いますので、このような質問をしています。

[toast-uz]

あと、DBは何でしょうか？DBによって非同期処理に対応しているかどうかがわかります。またDBによってはそもそも並列処理にも対応していない場合があるようです。

[python_heroku]

ご確認、ご質問ありがとうございます。

DBのI/O待ちネックを解消したいためになります。SQLによってリターンが1秒で返ってくるものもあれば、5分以上もかかるものが存在するためとなります。
10スレッド限定はDB事情であり、同時クエリ数ネックになります。
DBについてはHadoopとなります。

すいません。並列処理と非同期処理についてあまり区別はついてなく、少し調べて理解できなかったので、ご指摘、もしくは参考記事など教えていただけると非常にありがたいです。

[toast-uz]

コメントありがとうございます。状況はよくわかりました。確かにHadoopだとこういう状況になりますね。

並列処理と非同期の違いは
https://minowalab.org/python_parallel_multiprocess_noblocking_lib/
あたりを見ていただくとわかります。なお本家のドキュメントではここで言う並行処理を「スレッドベースの並列処理」と表現しています。並列と並行の用語の違いは諸説ありますので、気にしないでください。ポイントは「I/O待ちのブロッキングを避ける方法は非同期処理が推奨」であること。「その際はthreadingではなくasyncを使うこと」です。

本家はこちら
https://docs.python.org/ja/3/library/threading.html （スレッドベースの並列処理）
https://docs.python.org/ja/3/library/asyncio.html （非同期I/O）

よって非同期処理がオススメなのですが、DBアクセス用のPythonライブラリが非同期に対応しているか、ということも注意が必要です。「Python DB名 async」などでググって探しましょう。例えばmysqlだと信頼できる非同期ライブラリが無さそうですが、postgresqlだとありそうです。hadoopについては、ググるとpyhiveでasyncioが使えるようです。Hiveが使えるならこれがよいですね。
https://pypi.org/project/PyHive/
pyhiveのトップページのUsageによると、通常のAPIコールと非同期のAPIコールが両方用意されています。おそらく非同期版はクエリー発行後にすぐにPython側に制御が戻り、その後のwhileループで結果を取得するのでしょう。ですのて、whileで待ち受けつつ、最大10多重まで連続でクエリーを発行するような実装をすれば、質問者様の目的は達成しそうです。なお、先のリンクで紹介したPython本家のasyncioとは独立に実装されているようです。

なお、ここの質問は、スレッドベースの並列処理にアンカーされた質問のされかたをしてしまっていますので、いったんどれかにベストアンサーを出してクローズし、上記手段も検討された上で、再度詰まったところで質問を出されると、よいかと思います。

[python_heroku]

こちら非常に詳細にありがとうございます！
非同期についても次回何かするときに実装検討してみたいと思います！

Answer 1

concurrent.futuresパッケージのThreadPoolExecuterを使うといいと思います。

これでスレッドプールを10個にすれば、常時10個のスレッドを使って、終ったら終っていない処理をそのスレッドで実行するような動作ができます。

検索すればいくつも使いかたがでてきます。

Answer 2

ThreadPoolExecutorを使用する例です。
（データベースの詳細が書かれていないので何とも言えませんが、下記がうまく動くには、マルチスレッドで安全にアクセスできるデータベースであることが前提です）
（元の質問文と同様にきちんと動くコードではない部分（sql_list末尾の「~」等）が入っていますがそこは適宜読み替えてください。）

from concurrent.futures import ThreadPoolExecutor, as_completed

# sql引数に渡されたSQL文を実行する関数
def sql_excu(sql):
    result = ~~~~
    return result

# 同時実行する数
workers_count = 10
executor = ThreadPoolExecutor(max_workers=workers_count)

sql_list = ['sql1', 'sql2','sql3', ~]

futures = [executor.submit(sql_excu, sql) for sql in sql_list]

for future in as_completed(futures):
    print(future.result())

executor.shutdown()

Comments

[python_heroku]

こちらコードまでありがとうございます。
こちらの内容で実装したところ、うまく動作しました！

Answer 3

提示のコードではマルチスレッドで実行させてるようには見えませんが

Comments

[python_heroku]

はい。わかりにくくて申し訳ありません。
どうやって良いか全くわかっていないため、現状記載しているものでは全くマルチスレッドで動作になっていなかったり、SQLを何のDBに投げている、など全く分からないものになっておりますが、10個のスレッドで25個のSQLをなるべく早く処理させたい、ということを伝えられればと思い、あまり参考にもならないかもしれませんが、記載してみました。