サイトアイコン APPSWINGBY Developers Blog

golangの非同期処理にDispatcher-Workeパターンを利用してGraceful Shutdownを行う方法について

hidekazu kawashima

golangは、goroutine/channel を使って並行処理が簡単に実行できるので、重たい処理を非同期で実行しておくのは、非常に簡単です。ただし、並列処理するためのスレッドをいくらでも立てられるが、その立てスレッドが終わったのかを把握するためには、ちゃんと設計する必要があります。今回は、非同期処理で重たい処理を行っているプログラムがSIGTERMを受信した際に、安全に処理を継続して停止する仕組みについて、紹介します。

きっかけ

DockerAPI自体も終了時にSIGTERMを送ってくるので、そのSIGTERMハンドラを実装しないとイケないなということとで調べたのがきっかです。

Graceful shutdownとは

Graceful shutdownとは、一般的に以下の停止動作を指します。

例えば、バッチ処理や動画コンバート、DB更新などを伴う処理の場合は、処理中にアプリーケーションが中止してしまった場合、中途半端な状態に陥ってしまいます。結果の整合がとれる形でアプリーケーションが終了するようにしましょうというのが、Graceful shutdownです。

Dispatcher-Workerパターンとは

Dispatch-Workerパターンは、プロセス間通信で利用されるデザインパターンです。プロセス間通信では、タスク(やってほしい処理)を依頼するための、メッセージキューを経由して、別のworkerスレッドがタスクを処理するという形で実行します。Dispatcher-Workerパターンでは、そのWorkerへ指示を出す人にDispatcherという専用の人をおき、その処理で必要に応じて、空いているworkerへタスクを配るという形式です。図で表すと下記のような形になります。

間にDispacterがいることで、起動するWorkerスレッドの上限を制御できます。

この２つの処理を組み合わせたgo言語での実装例

goでは、goroutineという軽量スレッド、メッセージのやりとりにchannelが利用できます。下記にDispacherとWorkerの実装例を示します。全体は下記のような形です。

package main

import (
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

type (
    // Dispatcher
    Dispatcher struct {
        pool  chan *worker
        queue chan interface{}   // メッセージキュー
        workers []*worker
        wg      sync.WaitGroup
        quit    chan struct{}
    }

    // Worker
    worker struct {
        dispatcher *Dispatcher
        data chan interface{}
        quit chan struct{}
    }
)

func (d *Dispatcher) Start() {
    // ワーカを起動して処理まちの状態にしておく
    for _, w := range d.workers {
        w.start()
    }

    go func() {
        for {
            select {
                // メッセージがキューイングされた場合
            case v := 




メッセージを積むところからメッセージを取り出す流れ





利用者がメッセージを積む処理は、Addです。Addすることで、メッセージキュー(queue)へ書き込みが行われ、Dipatcherとして別途立てた別スレッド内で書き込みまちしている箇所(case v := <-d.queue)でメッセージを受信します。selectでは、queueに書き込まれるまでブロックされ、メッセージ受信するとメッセージを取り出します。





取り出しメッセージをworker配ってやってもらう流れ





現在処理を受け付けることができるworker(=IDLE状態)のものを取り出し(worker := <-d.pool)し、リクエスト内容をworkerへ送信(worker.data <- v)することで、worker側は、メッセージを受信(case v := <-w.data:)して、必要な処理を行います。





利用する側





下記のように必要なタイミングでDispacherのAddを呼ぶだけです。処理完了もWait()をコールするだけです。





    go func() {
        for i := 0; i 




SIGTERMを受け取る方法





Linuxのシステムプログラミング等でよく出てくるSIGTERMですが、Goでは、signal.Notifyで受け取るチャンネルを指定します。イベント発生するとそのチャンネル経由でイベント発生を検知できます。





SIGTERMのハンドリング、Dispatcher-Workerパターンを組み合わせると





package main

import (
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

type (
    // Dispatcher
    Dispatcher struct {
        pool  chan *worker
        queue chan interface{}   // メッセージキュー
        workers []*worker
        wg      sync.WaitGroup
        quit    chan struct{}
    }

    // Worker
    worker struct {
        dispatcher *Dispatcher
        data chan interface{}
        quit chan struct{}
    }
)

func (d *Dispatcher) Start() {
    // ワーカを起動して処理まちの状態にしておく
    for _, w := range d.workers {
        w.start()
    }

    go func() {
        for {
            select {
                // メッセージがキューイングされた場合
            case v := 




上の例では、SIGTERMを受信した際に、処理中の処理が完了するのを待っています。リクエストの部分は、このSIGTERMを受け取ったタイミング等で、ハンドラを解除するなどすることでリクエストも受け取らなくなるので、このような形でGraceful shutdownも実現できます。





おわりに





goでは、 goroutineでスレッド化が容易なことから安易な実装がされているケースもたまに見かけます。今回紹介したDispatcher-Workerパターンは、workerの数が制限できる為、スレッド数が無限に増えてしまう等の問題も回避できます。プロセス間通信を導入する際には、検討してみてはいかがでしょうか。
モバイルバージョンを終了