【Python】マルチスレッドを維持しつつ、それぞれのスレッドで定期処理を行いたい

Question

前提・実現したいこと

※Python学習を始めたばかりです。
マルチスレッドそれぞれで定期処理をするプログラムをいろいろ試してみたくて書いたソースです。
作成した内容では、suchedule_event_communicateとsuchedule_event_run_print_helloの
スレッドを立ち上げています。
それぞれで、3秒に一回時刻を表示、5秒に一回"HELLO!!!"を表示(ただし、処理の途中で10秒間待機させる)という内容になっています。

発生している問題・エラーメッセージ

下記のソースを実行すると、以下のメッセージのように、
同じ時刻で同じメッセージが二回表示されます。

2019-04-04 09:37:06.219 : get_time
2019-04-04 09:37:08.247 : HELLO!!!
2019-04-04 09:37:08.247 : HELLO!!!
2019-04-04 09:37:08.247 : 待機状態です、、、
2019-04-04 09:37:08.247 : 待機状態です、、、
2019-04-04 09:37:11.258 : 待機状態です、、、
2019-04-04 09:37:11.258 : 待機状態です、、、
2019-04-04 09:37:14.268 : 待機状態です、、、
2019-04-04 09:37:14.268 : 待機状態です、、、
2019-04-04 09:37:17.279 : 待機状態です、、、
2019-04-04 09:37:17.279 : 待機状態です、、、
2019-04-04 09:37:20.290 : print_hello_exit
2019-04-04 09:37:20.290 : print_hello_exit
2019-04-04 09:37:21.304 : get_time
2019-04-04 09:37:21.304 : get_time
2019-04-04 09:37:24.346 : get_time
2019-04-04 09:37:25.360 : HELLO!!!
2019-04-04 09:37:25.360 : 待機状態です、、、
2019-04-04 09:37:25.360 : HELLO!!!
2019-04-04 09:37:25.360 : 待機状態です、、、

該当のソースコード

import schedule
import time
import concurrent.futures

from datetime import datetime, timedelta


# ********************************
#   関数
# ********************************


def multi_thread():
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
    executor.submit(suchedule_event_communicate)
    executor.submit(suchedule_event_run_print_hello)


def get_timestring():
    """
    ミリ秒までを保持できる現在時刻を取得
    :return:
    """
    now = datetime.now()
    return now.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]


def print_message(run_time=get_timestring(), message=""):
    print(run_time + " : " + message)


def suchedule_event_communicate():
    """
    3秒に一回時刻を表示する
    :return:
    """
    schedule.every(3/60).minutes.do(print_time)
    while True:
        schedule.run_pending()
        time.sleep(1)


def print_time():
    print_message(get_timestring(), "get_time")


def suchedule_event_run_print_hello():
    """
    5秒に一回HELLO!!!を表示する
    :return:
    """
    schedule.every(5/60).minutes.do(print_hello)
    while True:
        schedule.run_pending()
        time.sleep(1)


def print_hello():
    now_time = get_timestring()
    print_message(now_time, "HELLO!!!")

    # HELLO!!!が表示されてから10秒間は待機時間
    while str(datetime.strptime(now_time, '%Y-%m-%d %H:%M:%S.%f') + timedelta(
            seconds=10)) > get_timestring():
        print_message(get_timestring(), "待機状態です、、、")
        time.sleep(3)

    print_message(get_timestring(), "print_hello_exit")


# ********************************
#   main
# ********************************


# 定期処理起動
multi_thread()

試したこと

マルチスレッドでwhileを使用する以下の処理であれば交互にメッセージを出力(同じ時刻・メッセージなし)しました。
scheduleの中で、whileや待機(sleep)と組み合わせるのが良くないのでしょうか。

import time
import concurrent.futures
from datetime import datetime


def get_timestring():
    """
    ミリ秒までを保持できる現在時刻を取得
    :return:
    """
    now = datetime.now()
    return now.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]


def print_message(run_time=get_timestring(), message=""):
    print(run_time + " : " + message)


def func1():
    while True:
        print_message(get_timestring(), "func1")
        time.sleep(1)


def func2():
    while True:
        print_message(get_timestring(), "func2")
        time.sleep(1)


if __name__ == "__main__":
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
    executor.submit(func1)
    executor.submit(func2)

試したこと②

suchedule_event_run_print_hello内に記載されていたwhileをなくしました。
そうするとうまくいくようです。

import schedule
import time
import concurrent.futures

from datetime import datetime, timedelta


# ********************************
#   関数
# ********************************


def multi_thread():
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
    executor.submit(suchedule_event_communicate)
    executor.submit(suchedule_event_run_print_hello)


def get_timestring():
    """
    ミリ秒までを保持できる現在時刻を取得
    :return:
    """
    now = datetime.now()
    return now.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]


def print_message(run_time=get_timestring(), message=""):
    print(run_time + " : " + message)


def suchedule_event_communicate():
    """
    3秒に一回時刻を表示する
    :return:
    """
    schedule.every(3/60).minutes.do(print_time)
    while True:
        schedule.run_pending()
        time.sleep(1)


def print_time():
    print_message(get_timestring(), "get_time")


def suchedule_event_run_print_hello():
    """
    5秒に一回HELLO!!!を表示する
    :return:
    """
    schedule.every(5/60).minutes.do(print_hello)
　　# whileをなくしました。
    schedule.run_pending()
    time.sleep(1)


def print_hello():
    now_time = get_timestring()
    print_message(now_time, "HELLO!!!")

    # HELLO!!!が表示されてから10秒間は待機時間
    while str(datetime.strptime(now_time, '%Y-%m-%d %H:%M:%S.%f') + timedelta(
            seconds=10)) > get_timestring():
        print_message(get_timestring(), "待機状態です、、、")
        time.sleep(3)

    print_message(get_timestring(), "print_hello_exit")


# ********************************
#   main
# ********************************


# 定期処理起動
multi_thread()

2019-04-05 10:00:37.653 : get_time
2019-04-05 10:00:39.653 : HELLO!!!
2019-04-05 10:00:39.661 : 待機状態です、、、
2019-04-05 10:00:42.661 : 待機状態です、、、
2019-04-05 10:00:45.661 : 待機状態です、、、
2019-04-05 10:00:48.662 : 待機状態です、、、
2019-04-05 10:00:51.662 : print_hello_exit
2019-04-05 10:00:52.662 : get_time
2019-04-05 10:00:55.662 : get_time
2019-04-05 10:00:56.662 : HELLO!!!
2019-04-05 10:00:56.663 : 待機状態です、、、
2019-04-05 10:00:59.663 : 待機状態です、、、

補足情報（FW/ツールのバージョンなど）

Python 3.7.0(pycharm使用)
windows7 64bit

Answer 1

質問のコードにあるscheduleはglobal変数です。そのためscheduleには２つのjobが定義されてしまい、それが２つのthreadで動作しているので、二重に実行されてしまいます。
その問題を解消するためには以下のように thread 毎にクラスSchedulerのインスタンスを作ればいいと思われます。

def suchedule_event_communicate():
    sc1 = schedule.Scheduler() 
    sc1.every(3/60).minutes.do(print_time)
    while True:
        sc1.run_pending()
        time.sleep(1)

def suchedule_event_run_print_hello():
    sc2 = schedule.Scheduler()
    sc2.every(5/60).minutes.do(print_hello)
    while True:
        sc2.run_pending()
        time.sleep(1)

Comments

[dodox86]

完全にかぶりました。失礼しました。

[_ng.h.ks_]

先に投稿いただけておりましたので、ベストアンサーにさせていただきました。
ご回答くださいましてありがとうございました。

Answer 2

原因としてはscheduleモジュールの関数版のschedule.everyとschedule.run_pendingを複数のスレッドで使っているためです。

scheduleモジュールのschedule.every関数を実行するとき、スケジューラーとして使用されるscheduleモジュール内の変数default_schedulerがグローバルとなっており、各スレッドでschedule.everyを使うと、それぞれのスレッドでひとつのスケジューラーを共用するかたちになります。

github schedule - init.py

スレッドそれぞれでスケジューラーを実行するため、同じ文字列が表示されてしまいます。質問者さんのコードではconcurrent.futures.ThreadPoolExecutorで生成したスレッドプールのスレッド2つが動きますが、それぞれで同じスケジュールを実行してしまうため、2つの"HELLO!!!"文字列がほぼ同時に表示されます。同じ時刻に見えるのはたまたまです。スレッドを3つにすると、今度は3つの"HELLO!!!"文字列が確認できます。

スレッドそれぞれでJOBをスケジューリングするのであれば、schedule.everyではなく、Scheduleクラスのインスタンスメソッドのeveryを使わなければなりません。また、schedule.run_pendingも同様です。上記で示した scheduleモジュールのソースファイル「init.py」を読むと分かります。

本件とは直接関係ありませんが、確認の際はthreading.get_identでスレッドのIDを表示させるようにすると分かり易くなります。例えばprint_messageに
適用すると、以下のようなかんじになります。

import threading

def print_message(run_time=get_timestring(), message=""):
    #print(run_time + " : " + message)
    print('{} (id={}): {}'.format(run_time, threading.get_ident(), message))

一部抜粋になりますが、質問者さんのコードをもとにして、以下のように修正します。

import threading

...

def print_message(run_time=get_timestring(), message=""):
    #print(run_time + " : " + message)
    print('{} (id={}): {}'.format(run_time, threading.get_ident(), message))


def suchedule_event_communicate():
    """
    3秒に一回時刻を表示する
    :return:
    """
    # schedule.everyではなく、インスタンスメソッドのeveryを使う
    scheduler = schedule.Scheduler()
    scheduler.every(3/60).minutes.do(print_time)
    while True:
        # schedule.run_pending ではなく、インスタンスメソッドのrun_pendingを使う
        scheduler.run_pending()
        time.sleep(1)

このようにすると、以下のように質問者さんの意図通り実行されます。

$ python3  t5.py
2019-04-05 16:25:20.673 (id=-2146563000): get_time
2019-04-05 16:25:22.674 (id=-2146563704): HELLO!!!
2019-04-05 16:25:22.689 (id=-2146563704): 待機状態です、、、
2019-04-05 16:25:23.673 (id=-2146563000): get_time
2019-04-05 16:25:25.690 (id=-2146563704): 待機状態です、、、
2019-04-05 16:25:26.673 (id=-2146563000): get_time
2019-04-05 16:25:28.691 (id=-2146563704): 待機状態です、、、
2019-04-05 16:25:29.674 (id=-2146563000): get_time
2019-04-05 16:25:31.691 (id=-2146563704): 待機状態です、、、
2019-04-05 16:25:32.674 (id=-2146563000): get_time
2019-04-05 16:25:34.691 (id=-2146563704): print_hello_exit
2019-04-05 16:25:35.674 (id=-2146563000): get_time
2019-04-05 16:25:38.674 (id=-2146563000): get_time
2019-04-05 16:25:39.691 (id=-2146563704): HELLO!!!

本件とは関係ないですが少し気になったところとしては、コードの一番下のメインスレッドでの実行は、すぐ終わってしまうことです。スレッドプールの2つのスレッドが起動したままなので気が付きにくいですが、メインは終わってしまってます。本来であれば何かしらThread.joinのようなもので待ち受けるべきだと思いますが、質問者さんの実験および習作のコードであろうし、別の課題として置いておきます。

Comments

[_ng.h.ks_]

細かくご説明いただいてありがとうございました。
参考になりました。
本件以外のところでも、指摘をしていただけまして勉強になりました。
もう少し勉強してみます。
ご回答くださいましてありがとうございました。