使用golang实现一个MapReduce的示例代码
作者：写代码的lorre
这篇文章主要给大家介绍了关于如何使用golang实现一个MapReduce,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
背景
在日常业务开发中，我们经常遇到需要并发处理的场景。例如：
- 依据id列表查询db，获取数据。为了保证查询性能，单次查询的id列表长度最好不要超过50(依据业务来判断)，当id列表长度超过50时，拆分成并发请求，减少耗时和提高性能，返回聚合后的结果
- 外部提供的接口不支持批量写入/读取数据，当需要批量处理数据时，为了减少耗时和提高性能，并发请求外部接口
以上处理数据的场景，都可以分成两个阶段：
- 请求阶段。基本都是IO操作，请求db，或者是调用外部接口
- 处理阶段。对返回的数据进行转换，过滤，聚合等操作
同步调用，调用耗时增长明显
并发调用，可以减少调用耗时
分析
上面说的处理数据的场景，都可以分成两个阶段：
- 请求阶段。IO操作，可以并发的去进行，互不干扰
- 处理阶段。同步进行，保证聚合结果的正确性
这种是一种特殊的MapReduce
为了处理这类场景，我们需要明确以下几个部分：
- 列表长度。代表有多少数据需要进行处理
- map函数。并发处理的函数，互不干扰
- reduce函数。同步处理的函数
- 最大并发数。决定需要开多少线程/协程来处理
- 拆分长度。列表长度 / 拆分长度 = 子任务数
由于我在日常开发中常使用golang语言，下面梳理下使用golang来解决这类问题的一个思路
函数签名
func ChunkProcess(length int, procedure func(start, end int) (interface{}, error),
reduce func(partialResult interface{}, partialErr error, start, end int), maxConcurrent int, chunkSize int) 核心逻辑：
当最大并发数 <= 1 或者子任务数(列表长度 / 拆分长度) <= 1时，同步执行map函数和reduce函数即可
其余情况，并发处理map函数，同步执行reduce函数
- 获取并发处理的子任务数量：lengthTask := int(math.Ceil(float64(length) / float64(chunkSize)))
- 通过sync.Mutex保证reduce同步执行
- 通过sync.WaitGroup保证等待子任务全部执行完成
- 通过chan控制最大并发数
代码实现
package test
import (
"math"
"sync"
)
func ChunkProcess(length int, procedure func(start, end int) (interface{}, error),
reduce func(partialResult interface{}, partialErr error, start, end int), maxConcurrent int, chunkSize int) {
if length < 1 {
return
}
if maxConcurrent <= 1 || length <= chunkSize {
doChunkProcessSerially(length, procedure, reduce, chunkSize)
} else {
doChunkProcessConcurrently(length, procedure, reduce, maxConcurrent, chunkSize)
}
}
// 同步处理
func doChunkProcessSerially(length int, procedure func(start, end int) (interface{}, error),
reduce func(partialResult interface{}, partialErr error, start, end int), chunkSize int) {
// 拆分的子任务数
chunkNums := int(math.Ceil(float64(length) / float64(chunkSize)))
for i := 0; i < chunkNums; i++ {
func(chunkIndex int) {
defer func() {
if err := recover(); err != nil {
// 自定义错误处理
}
}()
start := chunkIndex * chunkSize
end := start + chunkSize
if end > length {
end = length
}
// 执行map
response, err := procedure(start, end)
// 执行reduce
if reduce != nil {
reduce(response, err, start, end)
}
}(i)
}
}
// 并发处理
func doChunkProcessConcurrently(length int, procedure func(start, end int) (interface{}, error),
reduce func(partialResult interface{}, partialErr error, start, end int), maxConcurrent int, chunkSize int) {
index := 0
chunkIndex := 0
// 拆分的子任务数
lengthTask := int(math.Ceil(float64(length) / float64(chunkSize)))
// 保证reduce同步执行
var lock sync.Mutex
// 保证子任务全部执行完成
var wg sync.WaitGroup
wg.Add(lengthTask)
// 控制并发数
throttleChan := make(chan struct{}, maxConcurrent)
for {
start := index
end := index + chunkSize
if end > length {
end = length
}
throttleChan <- struct{}{}
go func(chunkIndex int) {
defer func() {
<-throttleChan
if err := recover(); err != nil {
// 自定义错误处理
}
wg.Done()
}()
// 执行map
response, err := procedure(start, end)
// 执行reduce
if reduce != nil {
lock.Lock()
defer lock.Unlock()
reduce(response, err, start, end)
}
}(chunkIndex)
chunkIndex++
index = index + chunkSize
if index >= length {
break
}
}
wg.Wait()
close(throttleChan)
}测试：
func TestChunkProcess(t *testing.T) {
trackIDs := []int64{123, 456, 789}
results := make([]int64, 0)
ChunkProcess(len(trackIDs), func(start, end int) (interface{}, error) {
result := trackIDs[start] + 100
return result, nil
}, func(partialResult interface{}, partialErr error, start, end int) {
results = append(results, partialResult.(int64))
}, 2, 1)
fmt.Println(results)
}总结
多对业务场景进行抽象分析，为这一类场景提供解决方案
以上就是使用golang实现一个MapReduce的详细内容，更多关于golang实现MapReduce的资料请关注脚本之家其它相关文章！