Go语言：完全深入理解数据并行和函数式 Map 1

大家好，我是一名 Go 语言工程师。我平日里也在教课。前几天给了一节关于 Go 语言数据并行的讲座。我在这里整理成文。希望大家喜欢。

从 Map、Reduce 等函数式编程讲起

在开发中，我们经常会将一个某种类型 T 的序列转化为类型 T2 的序列。最原始的方法就是使用 for loop。不过，for loop 不仅拗口，而且如果要做到良好的异步或者并发处理，就要写很多重复的代码。

所以，在各个编程语言中，标准库都提供了接口稳定的 map, reduce, filter 等函数。即使连 JavaScript 社区也开发了 RxJS 这样的函数式编程类库。

然而，Go 语言在标准可层面并没有这样的类库。作为上层业务逻辑开发者，这是不太方便的。所以社区内也有人尝试写出好的 Map、Reduce、Filter 等类库。一个最大的动力和需求就源于 Go 语言极度擅长并发编程。而 Map、Filter、Fold这三个函数是完全可以数据并行的（Reduce 不太行，我们之后会讲）。在业界用 Go 语言来做 Data Pipeline 其实也是很好的方案。

不过，写出一个好的 Map、Reduce 类库是需要考虑很多设计问题的。本文旨在给大家详尽并且深入地讲解这个问题。所有地代码都可以在 github.com/CreatCodeBu… 的数据并行（data concurrency）目录下找到。

如何写一个 Map

让我们从最基本的 for loop 开始

func Map(data []int, mapper func(int) int) []int {
	results := make([]int, len(data))
	for i, ele := range data {
		results[i] = mapper(ele)
	}
	return results
}
复制代码

这是一个最简单的 Map 实现。将一个 []int Map 到另外一个 []int 中。用法就是

func TestMap(t *testing.T) {
	results := Map([]int{1, 2, 3}, func(x int) int { return x + 1 })
	require.Equal(t, []int{2, 3, 4}, results)
}
复制代码

如你所见， 1, 2, 3 变成了 2, 3, 4 , 因为我们的 lambda 为 x + 1 。

然而，这里这个实现有两个问题。

第一，我们定死了原始类型和目标类型。如果我们想从 int Map 到 string，就要新写一个函数。

Go 没有泛型，这是造成这个原因的其中一点。不过，这只是开发时中的问题。比如 C++ 有基于代码生成模板的泛型，那么其实是在编译时生成更多的、属于不同类型的代码。所以，不管你是手写、还是编译器生成，在运行时代码都是一样多的。当然，如果像 Java 那样，通过运行时类型检查来实现泛型、就是另外一回事了。

所以，我认为这顶多叫做麻烦，而不是问题。

第二，更大的问题来自运行时

这个实现最大的问题，就是定死了原始数据序列和目标数据序列的内存模型。为什么必须要是 Slice 这种数据结构呢？Slice 所带来的一个副作用就是，数据最终会用 Array 存起来。而 Array 是连续的内存。然而，Map 这个函数从逻辑上根本没有要求数据要在物理上连续。Map 甚至都没有要求数据在逻辑上是连续的。

我为甚么不能从一个 Slice 映射到一个队列呢？为什么不能从一个 channel 映射到一个文件流呢？Map 的本质就是抽象的序列（数据流）之间的映射。所以，我们的实现应该表现出这一点，并且同时不要带来内存连续这样的副作用。我不是说内存连续不好，而是说不必要。 优秀的设计反应事物的本质。优秀的实现没有不必要的东西。

一个更好的 Map

先解决第二个问题

// producer 是一个数据生产者。Next 会迭代并返回序列中的下一个元素。
// 返回 io.EOF 表示穷尽了序列。其他错误值表示 producer 本身遇到了错误。
type producer interface {
	Next() (string, error)
}

// consumer 是数据消费者。Send 会读入新的数据。
type consumer interface {
	Send(int64)
}

// 返回错误如果 string 不能表示 int。比如 "xxx" 不是一个正确的 int 表示形式。
type mapper func(string) (int64, error)

func BetterMap(p producer, c consumer, mapper mapper) error {
	for {
		next, err := p.Next()
		if err != nil {
			if err == io.EOF {
				break
			} else {
				return err // 生产者本身遇到错误，终止 Map。
			}
		}
		datum, err := mapper(next)
		if err != nil {
			return err // mapper 出了问题，终止 Map。
		}
		c.Send(datum)
	}
	return nil
}

type StringProducer struct {
	index int
	data  []string
}

func (ip *StringProducer) Next() (string, error) {
	if ip.index < len(ip.data) {
		defer func() { ip.index++ }()
		return ip.data[ip.index], nil
	}
	return "", io.EOF
}

type OutputConsumer struct{}

func (c OutputConsumer) Send(ele int64) {
	fmt.Println(ele)
}
复制代码

我们完成了一个很大的提升，将数据的生产者和消费者的具体实现交给了 Map 的调用者，而不是 Map 自己来定义。Map 只定义 2 个大家都同意的接口。

用起来也非常方便

func ExampleBetterMap() {
	BetterMap(&StringProducer{data: []string{"1", "10", "11"}}, OutputConsumer{}, func(str string) (int64, error) {
		// 这里的 lambda 将字符串以二进制形式转为整数
		return strconv.ParseInt(str, 2, 64)
	})
	// Output: 1
	// 2
	// 3
}
复制代码

如你所见，我们可以随意地使用我们自己的实现。 consumer 甚至可以将结果 IO 出去，而不是存在内存里。这样 Map 函数就没有影响程序的内存效率。调用者代码和 Map 自己的权责分明了。同样的道理， producer 也可以将数据从其他源流读进来，而不是一次性地全部存在自己内部。