使用parallel-fn实现Golang高效并行执行

parallel-fn是一个轻量级的Golang库，专门用于简化并行函数执行。下面我将详细介绍其使用方法并提供示例代码。

安装

go get github.com/rafaeljesus/parallel-fn

基本用法

并行执行无返回值函数

package main

import (
	"fmt"
	"time"

	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	// 定义要并行执行的函数
	fn1 := func() error {
		time.Sleep(1 * time.Second)
		fmt.Println("Function 1 executed")
		return nil
	}

	fn2 := func() error {
		time.Sleep(2 * time.Second)
		fmt.Println("Function 2 executed")
		return nil
	}

	// 并行执行
	err := parallelfn.Run(fn1, fn2)
	if err != nil {
		fmt.Println("Error:", err)
	}

	fmt.Println("All functions completed")
}

并行执行有返回值函数

package main

import (
	"fmt"
	"time"

	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	// 定义有返回值的函数
	fn1 := func() (interface{}, error) {
		time.Sleep(1 * time.Second)
		return "Result from function 1", nil
	}

	fn2 := func() (interface{}, error) {
		time.Sleep(2 * time.Second)
		return 42, nil
	}

	// 并行执行并获取结果
	results, err := parallelfn.Exec(fn1, fn2)
	if err != nil {
		fmt.Println("Error:", err)
		return
	}

	// 处理结果
	for i, result := range results {
		fmt.Printf("Result %d: %v\n", i+1, result)
	}
}

高级特性

带超时控制的并行执行

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond)
	defer cancel()

	fn1 := func() (interface{}, error) {
		time.Sleep(1 * time.Second)
		return "Result 1", nil
	}

	fn2 := func() (interface{}, error) {
		time.Sleep(2 * time.Second) // 这个会超时
		return "Result 2", nil
	}

	results, err := parallelfn.ExecWithContext(ctx, fn1, fn2)
	if err != nil {
		fmt.Println("Error:", err) // 会输出超时错误
	}

	fmt.Println("Results:", results) // 只有第一个函数的结果
}

限制并发数

package main

import (
	"fmt"
	"time"

	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	// 创建并发限制为2的runner
	runner := parallelfn.NewRunner(parallelfn.WithMaxConcurrency(2))

	// 定义5个任务
	tasks := make([]func() error, 5)
	for i := 0; i < 5; i++ {
		i := i // 创建闭包变量
		tasks[i] = func() error {
			time.Sleep(1 * time.Second)
			fmt.Printf("Task %d completed\n", i+1)
			return nil
		}
	}

	// 执行，每次最多2个并发
	err := runner.Run(tasks...)
	if err != nil {
		fmt.Println("Error:", err)
	}
}

错误处理

parallel-fn会收集所有函数的错误并返回一个聚合错误：

package main

import (
	"errors"
	"fmt"

	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	fn1 := func() error {
		return errors.New("error from function 1")
	}

	fn2 := func() error {
		return errors.New("error from function 2")
	}

	err := parallelfn.Run(fn1, fn2)
	if err != nil {
		if multiErr, ok := err.(parallelfn.MultiError); ok {
			for _, e := range multiErr.Errors() {
				fmt.Println("Error:", e)
			}
		}
	}
}

性能考虑

1. 对于CPU密集型任务，并行数不要超过GOMAXPROCS
2. IO密集型任务可以适当增加并行数
3. 小任务考虑批量处理减少goroutine创建开销

parallel-fn是一个简单但强大的工具，适合需要并行执行多个独立任务的场景。对于更复杂的并行模式，可能需要考虑使用sync.WaitGroup或其他并发原语。