Python スレッドのフラグ管理をEnumとStateパターンでおこなう

はじめに

今までスレッドの状態管理をBool値を使っていたし状態遷移を直接代入方式でやっていた。今回、開発中のアプリのコード数が増えてきたこともあり、また、ワンランク上のコーディングをしたいという欲もでてきたので挑戦してみた。

以前のコード

class MainThread():
    def __init__(self) -> None:
        self.is_running_thread1 = True
        self.is_running_thread2 = True
        self.is_running_thread3 = True

    def task1(self):
        # スレッド1の状態がRUNか否か
        if self.is_running_thread1:
            print('thread1 is running.')
        time.sleep(1)
        # スレッド1シャットダウン
        self.is_running_thread1 = False
        # スレッド1の状態がシャットダウンか否か
        if not self.is_running_thread1:
            print('thread1 has shutdown.')
    
    # 以下省略
    ...
    ... 
    ...

コード数は短いけどテストがやりにくいし、なんか幼稚なコード。

挑戦したコード

# ---------
# スレッド名
# ---------

class ThreadName(Enum):
    THREAD1 = auto()
    THREAD2 = auto()
    THREAD3 = auto()

# -------------
# スレッドの状態
# -------------

class Thread1(Enum):
    RUN = auto()
    SHUTDOWN = auto()


class Thread2(Enum):
    RUN = auto()
    SHUTDOWN = auto()


class Thread3(Enum):
    RUN = auto()
    SHUTDOWN = auto()

# -------------
# Stateパターン
# -------------

class State():
    def __init__(self) -> None:
        self.thread1 = Thread1.RUN
        self.thread2 = Thread2.RUN
        self.thread3 = Thread3.RUN

    def shutdown(self, thread):
        if thread == ThreadName.THREAD1:
            self.thread1 = Thread1.SHUTDOWN
        elif thread == ThreadName.THREAD2:
            self.thread2 = Thread2.SHUTDOWN
        elif thread == ThreadName.THREAD3:
            self.thread3 = Thread3.SHUTDOWN

   # もしくはスレッドごとに分ける
   # そうすれば引数なしで呼べる
   # def shutdown_thread1,thread2,thread3

# --------------------------------
# スレッドで実行させるタスク
# 
# -- 行うこと --
# スレッドの状態がRUNか否か
# スレッドをシャットダウン
# スレッドの状態がシャットダウンか否か
# ---------------------------------

def task1(state: State):
    # スレッド1の状態がRUNか否か
    if state.thread1 == Thread1.RUN:
        print('thread1 is running.')
    time.sleep(1)
    # スレッド1をシャットダウン
    state.shutdown(ThreadName.THREAD1)
    # スレッド1の状態がシャットダウンか否か
    if state.thread1 == Thread1.SHUTDOWN:
        print('thread1 has shutdown.')

# 以下、task1と同様

def task2(state: State):
    if state.thread2 == Thread2.RUN:
        print('thread2 is running.')
    time.sleep(2)
    state.shutdown(ThreadName.THREAD2)
    if state.thread2 == Thread2.SHUTDOWN:
        print('thread2 has shutdown.')


def task3(state: State):
    if state.thread3 == Thread3.RUN:
        print('thread3 is running.')
    time.sleep(3)
    state.shutdown(ThreadName.THREAD3)
    if state.thread3 == Thread3.SHUTDOWN:
        print('thread3 has shutdown.')


def run(self):
    thread = MainThread()
    thread.run()


if __name__ == '__main__':
    # run()
    state = State()
    th1 = threading.Thread(target=task1, args=(state,))
    th2 = threading.Thread(target=task2, args=(state,))
    th3 = threading.Thread(target=task3, args=(state,))
    th1.start()
    th2.start()
    th3.start()
    th1.join()
    th2.join()
    th3.join()

感想

どうでしょうか

参考

https://nishi2.info/pydp/GoF_dp/behavior/20_State/index.html
https://ufcpp.net/study/csharp/st_enum.html