goのchannelを使って失敗してもリトライする処理を書いてみた

ある処理を行うデーモンを何個か作って並列に実行し
あるデーモンでエラーが起きたら再起動させる処理を書いてみました。

traceしやすいので
Daemon interface に自身のIDを出せるように設定しました。

interfaceとgoのchannelは便利だなぁと実感できるかと思います。

playground

package main

import (
    "context"
    "errors"
    "fmt"
    "sync"
    "time"
)

type Daemon interface {
    Do(ctx context.Context) chan error
    ID() string
}

type Sample struct {
    interval int
    id       string
}

func (a *Sample) Do(ctx context.Context) chan error {
    errCh := make(chan error)
    go func() {
        time.Sleep(time.Duration(a.interval) * time.Second)
        fmt.Printf("%#v, %#v\n", "fire", a.ID())
        errCh <- errors.New("err!!!")
    }()
    return errCh
}

func (a *Sample) ID() string {
    return a.id
}

type ConnTable struct {
    list map[chan error]Daemon
}

func (a *ConnTable) Start(ctx context.Context, d []Daemon) chan error {
    errCh := make(chan error)
    for _, item := range d {
        ch := item.Do(ctx)
        a.list[ch] = item
    }

    go func() {
        for ch, daemon := range a.list {
            go func(noticeCh chan error, d Daemon) {
                for {
                    err := <-noticeCh
                    fmt.Printf("[%#v] receive Err %#v\n", d.ID(), err)
                    noticeCh = d.Do(ctx)
                }
            }(ch, daemon)

        }
    }()

    return errCh
}

func main() {
    ss := []Daemon{
        &Sample{
            interval: 1,
            id:       "1",
        },
        &Sample{
            interval: 2,
            id:       "2",
        },
    }
    ctx, cancel := context.WithCancel(context.Background())

    ct := &ConnTable{
        list: map[chan error]Daemon{},
    }

    ct.Start(ctx, ss)

    time.Sleep(10 * time.Second)
    cancel()
}

結果

$ go run connpool/main.go
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "2"
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
["2"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "2"
["2"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "2"
["2"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "2"
["2"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}
"fire", "1"
["1"] receive Err &errors.errorString{s:"err!!!"}

Todo
再起動時の処理をサーキットブレーカにすればもっとよくなるはず