Go中如何并发过滤追加切片并保序且无数据竞争？

问题背景 #

以下代码尝试并发过滤输入切片中的偶数，但存在数据竞争导致数据丢失、结果顺序混乱的问题：

input := []int{1, 2, 3, 4, 5, 6, 7, 8}

wg := sync.WaitGroup{}

var result []int

for i, num := range input {
    wg.Add(1)
    go func(num, i int) {
        if num%2 == 0 {
            result = append(result, num)
        }

        wg.Done()
    }(num, i)
}
wg.Wait()

fmt.Println(result)

需求是得到有序结果：2,4,6,8，且输入是动态无上限的，无法预知过滤后的数据量。

解决方案 #

方法一：基于索引的临时切片（推荐，无锁高效） #

利用输入的固定顺序，预先创建与输入等长的临时切片（用指针标记有效值），每个goroutine仅写入对应索引的位置，最后再收集有效结果：

package main

import (
	"fmt"
	"sync"
)

func main() {
	input := []int{1, 2, 3, 4, 5, 6, 7, 8}

	wg := sync.WaitGroup{}
	// 创建与输入长度一致的临时切片，用*int表示是否有有效值
	temp := make([]*int, len(input))

	for i, num := range input {
		wg.Add(1)
		// 传递当前循环的num和索引i
		go func(val, idx int) {
			defer wg.Done()
			if val%2 == 0 {
				// 每个索引仅由一个goroutine写入，无数据竞争
				temp[idx] = &val
			}
		}(num, i)
	}
	wg.Wait()

	// 遍历临时切片，收集有效结果
	var result []int
	for _, ptr := range temp {
		if ptr != nil {
			result = append(result, *ptr)
		}
	}

	fmt.Println(result) // 输出：[2 4 6 8]
}

优势：无锁操作，完全保证顺序，性能最优，适合大部分场景。

方法二：带索引的通道接收（适合超大输入） #

若输入量极大，不想预先分配大内存，可通过通道传递带原始索引的结果，主线程按索引顺序整理：

package main

import (
	"fmt"
	"sync"
)

func main() {
	input := []int{1, 2, 3, 4, 5, 6, 7, 8}

	wg := sync.WaitGroup{}
	// 通道缓冲设为输入长度，避免goroutine阻塞
	ch := make(chan struct {
		idx int
		val int
	}, len(input))

	for i, num := range input {
		wg.Add(1)
		go func(val, idx int) {
			defer wg.Done()
			if val%2 == 0 {
				ch <- struct {
					idx int
					val int
				}{idx, val}
			}
		}(num, i)
	}

	// 等待所有goroutine完成后关闭通道
	go func() {
		wg.Wait()
		close(ch)
	}()

	// 初始化临时切片存储结果
	temp := make([]*int, len(input))
	for res := range ch {
		temp[res.idx] = &res.val
	}

	// 收集最终有序结果
	var result []int
	for _, ptr := range temp {
		if ptr != nil {
			result = append(result, *ptr)
		}
	}

	fmt.Println(result) // 输出：[2 4 6 8]
}

优势：无需一次性分配与输入等长的内存，适合超大批量数据处理，仍能保证顺序。

方法三：互斥锁保护切片（不推荐） #

用sync.Mutex保护append操作可解决数据丢失，但无法保证顺序，且会让并发操作退化为串行，丧失并发优势，仅作参考：

package main

import (
	"fmt"
	"sync"
)

func main() {
	input := []int{1, 2, 3, 4, 5, 6, 7, 8}

	wg := sync.WaitGroup{}
	var mu sync.Mutex
	var result []int

	for i, num := range input {
		wg.Add(1)
		go func(val, idx int) {
			defer wg.Done()
			if val%2 == 0 {
				mu.Lock()
				result = append(result, val)
				mu.Unlock()
			}
		}(num, i)
	}
	wg.Wait()

	// 结果顺序无法保证，例如可能输出[4 2 6 8]
	fmt.Println(result)
}

内容的提问来源于stack exchange，提问作者yssachan