Golang结构化并发编程简化插件库flowmatic的使用

Flowmatic是一个通用的Go库，提供了一种结构化的并发编程方法。它让你可以轻松管理并发任务，方式简单但有效且灵活。

特性

• 具有简单的API，比使用channel/waitgroup/mutex更具可读性
• 处理各种并发问题，如异构任务组、对切片执行同构任务以及动态工作生成
• 聚合错误
• 正确跨goroutine边界传播panic
• 提供上下文取消的帮助函数
• 依赖少
• 测试覆盖率高

如何使用Flowmatic

执行异构任务

Flowmatic解决了管理并行执行多个独立任务的问题。例如，你想向三个不同的下游API发送数据。如果任何发送失败，你想返回一个错误。使用传统的Go并发，这会很快变得复杂难以管理。Flowmatic让它变得简单。

使用flowmatic.Do执行异构任务：

err := flowmatic.Do(
    func() error {
        return doThingA(),
    },
    func() error {
        return doThingB(),
    },
    func() error {
        return doThingC(),
    })

要在第一个错误时取消任务的上下文，使用flowmatic.All。要在第一个成功时取消任务的上下文，使用flowmatic.Race。

// 创建变量保存响应
var pageA, pageB, pageC string
// 竞争请求看谁先响应
err := flowmatic.Race(ctx,
    func(ctx context.Context) error {
        var err error
        pageA, err = request(ctx, "A")
        return err
    },
    func(ctx context.Context) error {
        var err error
        pageB, err = request(ctx, "B")
        return err
    },
    func(ctx context.Context) error {
        var err error
        pageC, err = request(ctx, "C")
        return err
    },
)

执行同构任务

flowmatic.Each在需要对切片中的每个项目使用工作池执行相同任务时很有用：

things := []someType{thingA, thingB, thingC}

err := flowmatic.Each(numWorkers, things,
    func(thing someType) error {
        foo := thing.Frobincate()
        return foo.DoSomething()
    })

使用flowmatic.Map将输入切片映射到输出切片：

func Google(ctx context.Context, query string) ([]Result, error) {
    searches := []Search{Web, Image, Video}
    return flowmatic.Map(ctx, flowmatic.MaxProcs, searches,
        func(ctx context.Context, search Search) (Result, error) {
            return search(ctx, query)
        })
}

管理生成新任务的任务

对于可能创建更多工作的任务，使用flowmatic.ManageTasks。创建一个将被串行执行的manager，并让它保存结果并检查任务的输出以决定是否有更多工作要做。

// 任务获取页面并提取URL
task := func(u string) ([]string, error) {
    page, err := getURL(ctx, u)
    if err != nil {
        return nil, err
    }
    return getLinks(page), nil
}

// 从页面到链接的映射
// 不需要锁，因为只有manager会访问它
results := map[string][]string{}
var managerErr error

// manager跟踪哪些页面已被访问和结果图
manager := func(req string, links []string, err error) ([]string, bool) {
    // 在第一个错误后停止执行
    if err != nil {
        managerErr = err
        return nil, false
    }
    // 在map中保存最终结果
    results[req] = urls

    // 检查要抓取的新页面
    var newpages []string
    for _, link := range links {
        if _, ok := results[link]; ok {
            // 已见过，尝试下一个链接
            continue
        }
        // 添加到新页面列表
        newpages = append(newpages, link)
        // 向map添加占位符以防止重复抓取
        results[link] = nil
    }
    return newpages, true
}

// 使用与GOMAXPROCS一样多的工作者处理任务
flowmatic.ManageTasks(flowmatic.MaxProcs, task, manager, "http://example.com/")
// 检查是否有错误
if managerErr != nil {
    fmt.Println("error", managerErr)
}

使用TaskPool的高级模式

对于非常高级的用途，flowmatic.TaskPool消除了管理工作池的样板代码。

// MD5All读取根目录下的所有文件
// 并返回从文件路径到文件内容MD5总和的映射
// 如果目录遍历失败或任何读取操作失败，
// MD5All返回一个错误
func MD5All(ctx context.Context, root string) (map[string][md5.Size]byte, error) {
    // 创建20个digester的池
    in, out := flowmatic.TaskPool(20, digest)

    m := make(map[string][md5.Size]byte)
    // 打开两个goroutine:
    // 一个用于通过遍历文件系统读取文件名
    // 一个用于记录digester的结果到map中
    err := flowmatic.All(ctx,
        func(ctx context.Context) error {
            return walkFilesystem(ctx, root, in)
        },
        func(ctx context.Context) error {
            for r := range out {
                if r.Err != nil {
                    return r.Err
                }
                m[r.In] = *r.Out
            }
            return nil
        },
    )

    return m, err
}

关于panic的注意事项

在Go中，如果goroutine中出现panic且未被恢复，则整个进程将关闭。Flowmatic通过捕获工作goroutine中发生的panic并在父goroutine中重新传播它来解决这个问题，因此可以在适当的级别捕获和记录panic。