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
再起動時の処理をサーキットブレーカにすればもっとよくなるはず