Golang实现可中断执行流程的灵活机制插件库breaker的使用

💡 核心思想

breaker提供了一种灵活的机制来使执行流程可中断。它携带一个取消信号来中断动作执行。

var NewYear = time.Time{}.AddDate(time.Now().Year(), 0, 0)

interrupter := breaker.Multiplex(
    breaker.BreakByContext(context.WithTimeout(req.Context(), time.Minute)),
    breaker.BreakByDeadline(NewYear),
    breaker.BreakBySignal(os.Interrupt),
)
defer interrupter.Close()

<-interrupter.Done() // 等待上下文取消、超时或中断信号

🏆 使用动机

breaker可以与其他库如retry和semaphore配合使用，使它们更加一致和可靠。

// 与retry配合使用
if err := retry.Retry(breaker.BreakByTimeout(time.Minute), action); err != nil {
    log.Fatal(err)
}

// 与semaphore配合使用
if err := semaphore.Acquire(breaker.BreakByTimeout(time.Minute), 5); err != nil {
    log.Fatal(err)
}

🤼‍♂️ 使用示例

带重试的HTTP请求

interrupter := breaker.Multiplex(
    breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
    breaker.BreakByTimeout(timeout),
)
defer interrupter.Close()

ctx := breaker.ToContext(interrupter)
ctx = context.WithValue(ctx, header, "...")

req, err := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil)
if err != nil {
    panic(err)
}

var resp *http.Response
action := func(ctx context.Context) (err error) {
    req = req.Clone(ctx)

    source := ctx.Value(header).(string)
    req.Header.Set(header, source)

    resp, err = http.DefaultClient.Do(req)
    return err
}

if err := retry.Do(ctx, action); err != nil {
    panic(err)
}

完整示例：

package main

import (
    "context"
    "fmt"
    "io"
    "net/http"
    "net/http/httptest"
    "os"
    "syscall"
    "time"

    "github.com/kamilsk/breaker"
    "github.com/kamilsk/retry/v5"
)

func main() {
    const (
        header  = "X-Message"
        timeout = time.Minute
    )

    server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
        time.Sleep(timeout / 10)
        _, _ = rw.Write([]byte(req.Header.Get(header)))
    }))
    defer server.Close()

    interrupter := breaker.Multiplex(
        breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
        breaker.BreakByTimeout(timeout),
    )
    defer interrupter.Close()

    ctx := breaker.ToContext(interrupter)
    ctx = context.WithValue(ctx, header, "flexible mechanism to make execution flow interruptible")

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil)
    if err != nil {
        panic(err)
    }

    var resp *http.Response
    action := func(ctx context.Context) (err error) {
        req = req.Clone(ctx)

        source := ctx.Value(header).(string)
        req.Header.Set(header, source)

        resp, err = http.DefaultClient.Do(req)
        return err
    }

    if err := retry.Do(ctx, action); err != nil {
        fmt.Println("error:", err)
        return
    }
    _, _ = io.Copy(os.Stdout, resp.Body)
}

HTTP服务器的优雅关闭

interrupter := breaker.Multiplex(
    breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
    breaker.BreakByTimeout(timeout),
)
defer interrupter.Close()

server := http.Server{
    BaseContext: func(net.Listener) context.Context {
        return breaker.ToContext(interrupter)
    },
}
go func() {
    if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
        log.Fatal(err)
    }
}()

<-interrupter.Done()
if errors.Is(interrupter.Err(), breaker.Interrupted) {
    if err := server.Shutdown(context.TODO()); err != nil {
        panic(err)
    }
}

完整示例：

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "net"
    "net/http"
    "os"
    "syscall"
    "time"

    "github.com/kamilsk/breaker"
)

func main() {
    const timeout = time.Minute

    interrupter := breaker.Multiplex(
        breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
        breaker.BreakByTimeout(timeout),
    )
    defer interrupter.Close()

    server := http.Server{
        Addr:    ":8080",
        Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}),
        BaseContext: func(net.Listener) context.Context {
            return breaker.ToContext(interrupter)
        },
    }
    go func() {
        if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
            log.Fatal(err)
        }
    }()

    <-interrupter.Done()
    if err := interrupter.Err(); errors.Is(err, breaker.Interrupted) {
        if err := server.Shutdown(context.TODO()); err != nil {
            panic(err)
        }
    }
    fmt.Println("graceful shutdown")
}

🧩 集成安装

使用go modules管理版本：

$ go get github.com/kamilsk/breaker@latest

🤲 应用场景

控制台工具执行限时命令

示例展示了如何执行限时10分钟的console命令：

$ date
# Thu Jan  7 21:02:21
$ breakit after 10m -- server run --port=8080
$ breakit ps
# +--------------------------+----------------------------+----------+----------+
# | Process                  | Status                     | Since    | Until    |
# +--------------------------+----------------------------+----------+----------+
# | server run --port=8080   | exit 1; panic: database... | 21:02:36 | -        |
# +--------------------------+----------------------------+----------+----------+
# |                          |                            |    Total |        1 |
# +--------------------------+----------------------------+----------+----------+
$ breakit after 10m -- database run --port=5432
$ breakit after 10m delay 5s -- server run --port=8080
$ breakit ps
# +--------------------------+----------------------------+----------+----------+
# | Process                  | Status                     | Since    | Until    |
# +--------------------------+----------------------------+----------+----------+
# | database run --port=5432 | running                    | 21:04:09 | 21:14:09 |
# | server run --port=8080   | delayed                    | 21:04:30 | 21:14:25 |
# +--------------------------+----------------------------+----------+----------+
# |                          |                            |    Total |        2 |
# +--------------------------+----------------------------+----------+----------+