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也能够将数据从其余源流读进来，而不是一次性地所有存在本身内部。

再来解决第一个问题

不过，咱们仍然须要解决第一个问题：就是针对不一样类型的 Map。

咱们会在下文中讲解。

扫码关注哲的代码实验室