使用parapipe实现Golang并行FIFO管道处理

parapipe是一个用于Golang的并行处理库，它可以在保持消息顺序的同时实现并行处理。下面我将介绍如何使用parapipe库来实现并行FIFO管道处理。

parapipe基本概念

parapipe的核心思想是：

1. 并行处理消息以提高吞吐量
2. 保持消息的原始顺序
3. 提供简单的API接口

安装parapipe

首先安装parapipe库：

go get github.com/nazar256/parapipe

基本使用示例

package main

import (
	"fmt"
	"time"

	"github.com/nazar256/parapipe"
)

func main() {
	// 创建一个并行管道，设置工作协程数为4
	pipe := parapipe.New(4)

	// 启动管道
	pipe.Start()

	// 向管道发送数据
	for i := 1; i <= 10; i++ {
		num := i // 创建副本
		pipe.In() <- parapipe.NewJob(num, func() (interface{}, error) {
			// 模拟耗时处理
			time.Sleep(time.Second)
			return num * num, nil
		})
	}

	// 关闭输入通道
	close(pipe.In())

	// 从输出通道读取结果
	for result := range pipe.Out() {
		if result.Err != nil {
			fmt.Printf("处理错误: %v\n", result.Err)
			continue
		}
		fmt.Printf("输入: %d, 输出: %v\n", result.Value.(int), result.ProcessedValue)
	}
}

高级特性

1. 自定义错误处理

pipe := parapipe.NewWithConfig(parapipe.Config{
	WorkerCount: 4,
	OnError: func(err error) {
		fmt.Printf("捕获到错误: %v\n", err)
	},
})

2. 控制并发度

// 动态调整工作协程数
pipe.SetWorkerCount(8)

3. 带缓冲的管道

pipe := parapipe.NewWithConfig(parapipe.Config{
	WorkerCount: 4,
	BufferSize:  100, // 设置缓冲区大小
})

实际应用示例

假设我们需要并行处理一批URL请求，但需要保持响应顺序：

package main

import (
	"fmt"
	"io/ioutil"
	"net/http"
	"time"

	"github.com/nazar256/parapipe"
)

func main() {
	urls := []string{
		"https://example.com",
		"https://google.com",
		"https://github.com",
		"https://golang.org",
	}

	// 创建管道，2个并行工作者
	pipe := parapipe.New(2)
	pipe.Start()

	// 发送任务
	for i, url := range urls {
		idx := i
		u := url
		pipe.In() <- parapipe.NewJob(idx, func() (interface{}, error) {
			resp, err := http.Get(u)
			if err != nil {
				return nil, err
			}
			defer resp.Body.Close()
			
			body, err := ioutil.ReadAll(resp.Body)
			if err != nil {
				return nil, err
			}
			
			return len(body), nil
		})
	}

	close(pipe.In())

	// 接收结果
	for result := range pipe.Out() {
		if result.Err != nil {
			fmt.Printf("URL %d 处理失败: %v\n", result.Value.(int), result.Err)
			continue
		}
		fmt.Printf("URL %d 响应大小: %d bytes\n", result.Value.(int), result.ProcessedValue.(int))
	}
}

性能考虑

1. 工作协程数应根据任务类型和机器核心数调整
2. I/O密集型任务可以设置较多协程
3. CPU密集型任务协程数不宜过多

错误处理建议

1. 始终检查result.Err
2. 使用OnError回调进行统一错误处理
3. 考虑实现重试逻辑

parapipe提供了一种简单而强大的方式来并行处理数据流同时保持顺序，非常适合需要并行处理但要求结果顺序与输入顺序一致的应用场景。