多机分布式执行异步任务的实现姿势
通过多机分片执行异步任务的一种实现姿势，基于GO语言实现。
序言
执行异步任务时，比如需要处理10W个订单，如果是PHP，我们一般会配置一个定时任务，然后该定时任务就会在单机上执行；如果是GO或者JAVA，我们也需要使用相应的策略，保证该任务只在单机上执行，比如分布式锁。可能有同学会问，我直接在多机上执行同一个任务不行么，我只想说，你胆子真大，当多机同时处理一条数据，你会死的很惨的。
那我们是否有一种方法，可以让任务在多机同时执行，然后又可以避免多机同时处理相同数据的问题呢？这里给大家介绍一种多机分片的方式，也是最近在公司Get到的新技能。
应用场景
最近在做异步任务迁移，要求对DB中的订单进行处理，因为订单的数量非常大，10W的数量级是常规状态，如果只通过一台机器去处理，执行效率非常低，所以需要通过多机并发处理。
对于上述方式，其实还有另外一种解决方案，就是单机执行任务，然后把任务放入消息队列，再新增一个接口，用于消费队列中的数据，然后进行数据处理，因为接口对应的服务是集群部署，所以执行速度很快，不过这里在设计方案时，需要考虑消息重复消费，多机可能同时处理单条消息，网络异常导致消息未得到处理等问题，具体解决方案，欢迎大家线下和我讨论哈。
多机分片
什么是多机分片呢？说的通俗一点，就是把数据分成N份，分别给每一台机器执行。比如我们有1000条数据，通过相应策略，将数据分成5份，每份数据200条，如果我们有5台机器，那么每台机器可以分别处理200条数据。
那么具体是怎么实现？
为了更好讲解，我先简单模拟一下场景：
- DB包含20条数据，DB主键ID为0、1、2、3 ... 19；
- 有3台机器，每台机器起一个线程跑任务，共起3个线程；
- 需要将数据分成10份，每份数据有2条，然后分给这3个线程。
令牌获取
将数据分成10分，就有10个令牌，即number=10，分别为0、1、2 ... 9，处理逻辑如下：
- 每个任务有任务名，以任务名为key，通过increase=redis.incr(key)计数，然后将increase值通过number取模，得到令牌token=increase%number，第一次执行的increase=1，所以token=1%10=1；
- 构造令牌tokenKey=key+token=key1，然后通过redis对tokenKey加一个分布式锁，如果加锁成功，返回令牌值；
- 如果加锁失败，循环执行increase=redis.incr(key)，此时increase=2，token=2%10=2，拿到令牌tokenKey=key2，再执行分布式锁，成功返回，未成功，同上依次反复。
Redis Incr 命令将 key 中储存的数字值增一。如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作。
分片获取DB数据
机器的线程拿到令牌后，就可以去分片获取数据了，假如DB的数据结构如下，且只有20条数据：
| 订单号orderId | 商品名productName | 配送状态status |
|---|---|---|
| 0 | 数据线 | 0 |
| 1 | 键盘 | 0 |
| 2 | 显示器 | 0 |
| ... | ... | ... |
| 19 | 鼠标 | 0 |
下面看一下分片获取数据流程：
- 当线程拿到的令牌为token=0，可以通过select * from tableName wwhere orderId % 10 = token and status = 0 limit pageSize;（假如pageSize=100），因为取模匹配的数据的orderId=0和10，所以该线程可以拿到0和10这两条数据，如果pageSize=1，那就只能拿到0这条数据，数据10等下次处理时再获取；
- 拿到分片数据后，就可以开始对数据进行逻辑处理，处理完毕后，需要将status置为1，避免下次再扫描到该数据。
留给大家一个问题，如果有一条数据一直处理失败，每次获取数据，都会先获取到这条问题数据，那么有什么策略可以让这条数据推后执行呢？
实战模拟
这里需要用到分布式锁，分布式锁的代码，已经在文章“Redis实现分布式锁”中已经说明，下面先看获取释放令牌的代码：
const NO_INDEX = 100
const REDIS_ALIAS_NAME = "jointly"
const TASK_NAME = "task_name"
const RANGE_DATA = int64(10)
const PAGE_SIZE = int64(2)
// 分片任务
type FragmentTask struct {
RedisLock
}
// 获取令牌
func (f *FragmentTask) GetToken(processId int64) (tokenId int64, err error) {
i := 0
for {
increase, err := redis.Incr(REDIS_ALIAS_NAME, TASK_NAME)
if err != nil {
return 0, err
}
tokenId := increase % RANGE_DATA
lockKey := TASK_NAME + string(tokenId)
if f.GetDistributeLock(lockKey, 60) {
fmt.Printf("Get lock key success, processId:%d, tokenId:%d\n", processId, tokenId)
return tokenId, nil
}
fmt.Printf("Get lock key conflict, processId:%d, tokenId:%d\n", processId, tokenId)
i++
if int64(i) >= RANGE_DATA {
fmt.Printf("Begin a new cycle.\n")
return NO_INDEX, nil
}
}
}
// 释放令牌锁
func (f *FragmentTask) ReleaseToken(tokenId int64) bool {
lockKey := TASK_NAME + string(tokenId)
ret := f.DelDistributeLock(lockKey)
if !ret {
fmt.Printf("Release token failed, tokenId:%d\n", tokenId)
}
return ret
}令牌生成的流程，前面已经详细讲解，这里需要注意的是，我们每次只遍历rangData范围，超过该范围后会退出，外层其实有个循环，会重新进入。
我们再看看通过令牌获取分片数据的逻辑：
func ( *Order) QueryOrderList(rangeData, tokenId, pageSize int64) (data []OrderData, err error){
o := orm.NewOrm()
o.Using("default")
num, err := o.Raw("SELECT * from "+ "tb_order where status = 0 and order_id % ? = ? limit ?", rangeData, tokenId, pageSize).QueryRows(&data)
if err != nil {
return nil, err
}
if num > 0 {
}
return data, nil
}下面是单个线程的任务处理流程：
// 处理任务
func (f *FragmentTask) DoProcess(processId int64) error {
order := &db.Order{}
for {
tokenId, err := f.GetToken(processId)
if err != nil {
fmt.Printf("failed, exist!\n")
return err
}
// 所有的令牌都锁住了，睡眠以后，再重新执行
if tokenId == NO_INDEX {
fmt.Printf("All token is conflict, sleep for a while.\n")
time.Sleep(time.Second * 8)
continue
}
orderList, err := order.QueryOrderList(RANGE_DATA, tokenId, PAGE_SIZE)
if err != nil {
fmt.Printf("Query order list failed, tokenId:%d, err:%s\n", tokenId, err.Error())
f.ReleaseToken(tokenId)
continue
}
fmt.Printf("Begin to process, processId:%d, tokenId:%d, print orderList:%v\n", processId, tokenId, orderList)
// 处理任务，用sleep模拟
time.Sleep(time.Second * 1)
// 处理完数据，更新DB记录状态
for _, orderRecord := range orderList {
orderRecord.Status = 1
order.UpdateOrderStatus(&orderRecord)
}
f.ReleaseToken(tokenId)
}
return nil
}这个逻辑是不是很清晰，就是一个SQL查询。最后就是多线程处理逻辑，我们只开了3个线程，模拟3台机器（假如每台机器只有一个线程）：
// 测试任务分片
func FragmentTest(fragmentTask *redis.FragmentTask) {
// 开启3个线程（模拟3台机器），去处理任务
for i := 0; i <= 2; i ++ {
go fragmentTask.DoProcess()
}
// 避免子线程退出，主线程睡一会
time.Sleep(time.Second * 100)
}
func main() {
redisLock := &redis.RedisLock{}
order := &db.Order{}
fragmentTask := &redis.FragmentTask{}
// 初始化资源
redisLock.IntiRedis()
order.InitDb()
// 测试任务分片
FragmentTest(fragmentTask)
return
}我们先看看DB执行前数据，初始状态status都是0，然后order_id是主键：
mysql> select * from tb_order;
+----+----------+--------------+--------+
| id | order_id | product_name | status |
+----+----------+--------------+--------+
| 1 | 1 | 鼠标1 | 0 |
| 2 | 2 | 鼠标2 | 0 |
| 3 | 3 | 鼠标3 | 0 |
| 4 | 4 | 鼠标4 | 0 |
| 5 | 5 | 鼠标5 | 0 |
| 6 | 6 | 鼠标6 | 0 |
| 7 | 7 | 鼠标7 | 0 |
| 8 | 8 | 鼠标8 | 0 |
| 9 | 9 | 鼠标9 | 0 |
| 10 | 10 | 鼠标10 | 0 |
| 11 | 11 | 鼠标11 | 0 |
| 12 | 12 | 鼠标12 | 0 |
| 13 | 13 | 鼠标13 | 0 |
| 14 | 14 | 鼠标14 | 0 |
| 15 | 15 | 鼠标15 | 0 |
| 16 | 16 | 鼠标16 | 0 |
| 17 | 17 | 鼠标17 | 0 |
| 18 | 18 | 鼠标18 | 0 |
| 19 | 19 | 鼠标19 | 0 |
| 20 | 20 | 鼠标20 | 0 |
+----+----------+--------------+--------+直接看执行结果：
Get lock key success, processId:0, tokenId:1
Get lock key success, processId:1, tokenId:2
Get lock key success, processId:2, tokenId:3
Begin to process, processId:0, tokenId:1, print orderList:[{1 1 鼠标1 0} {11 11 鼠标11 0}]
Begin to process, processId:2, tokenId:3, print orderList:[{3 3 鼠标3 0} {13 13 鼠标13 0}]
Begin to process, processId:1, tokenId:2, print orderList:[{2 2 鼠标2 0} {12 12 鼠标12 0}]
Get lock key success, processId:0, tokenId:4
Begin to process, processId:0, tokenId:4, print orderList:[{4 4 鼠标4 0} {14 14 鼠标14 0}]
Get lock key success, processId:1, tokenId:5
Begin to process, processId:1, tokenId:5, print orderList:[{5 5 鼠标5 0} {15 15 鼠标15 0}]
Get lock key success, processId:2, tokenId:6
Begin to process, processId:2, tokenId:6, print orderList:[{6 6 鼠标6 0} {16 16 鼠标16 0}]
Get lock key success, processId:0, tokenId:7
Begin to process, processId:0, tokenId:7, print orderList:[{7 7 鼠标7 0} {17 17 鼠标17 0}]
Get lock key success, processId:1, tokenId:8
Begin to process, processId:1, tokenId:8, print orderList:[{8 8 鼠标8 0} {18 18 鼠标18 0}]
Get lock key success, processId:2, tokenId:9
Begin to process, processId:2, tokenId:9, print orderList:[{9 9 鼠标9 0} {19 19 鼠标19 0}]
Get lock key success, processId:0, tokenId:0
Begin to process, processId:0, tokenId:0, print orderList:[{10 10 鼠标10 0} {20 20 鼠标20 0}]
Get lock key success, processId:1, tokenId:1
Begin to process, processId:1, tokenId:1, print orderList:[]
Get lock key success, processId:2, tokenId:2
Begin to process, processId:2, tokenId:2, print orderList:[]
Get lock key success, processId:0, tokenId:3
Get lock key success, processId:1, tokenId:4
Begin to process, processId:1, tokenId:4, print orderList:[]
Get lock key success, processId:2, tokenId:5
Begin to process, processId:0, tokenId:3, print orderList:[]
Begin to process, processId:2, tokenId:5, print orderList:[]我们简单分析一下，我们每次从DB获取数据，设置的PageSize=2，所以每个线程每次从DB会获取2条数据，比如tokenId=7的线程，会从DB拿到[{7 7 鼠标7 0} {17 17 鼠标17 0}]这两条数据，这两条数据的order_id分别为7和17，因为我们的是把数据分成了10分，所以是通过10取的模，取模值和tokenId相等。通过上面的输出，我们可以很清晰看到，线程0拿到0、1、4、7这4个令牌，线程1拿到2、5、8这3个令牌，线程2拿到3、6、9这3个令牌，3个线程拿到的令牌互不冲突，最后从DB查询的数据也不会冲突。我们把PageSize设置为1，再看看执行效果：
Get lock key success, processId:0, tokenId:9
Get lock key success, processId:1, tokenId:0
Get lock key success, processId:2, tokenId:1
Begin to process, processId:0, tokenId:9, print orderList:[{9 9 鼠标9 0}]
Begin to process, processId:2, tokenId:1, print orderList:[{1 1 鼠标1 0}]
Begin to process, processId:1, tokenId:0, print orderList:[{10 10 鼠标10 0}]
Get lock key success, processId:0, tokenId:2
Begin to process, processId:0, tokenId:2, print orderList:[{2 2 鼠标2 0}]
Get lock key success, processId:1, tokenId:3
Begin to process, processId:1, tokenId:3, print orderList:[{3 3 鼠标3 0}]
Get lock key success, processId:2, tokenId:4
Begin to process, processId:2, tokenId:4, print orderList:[{4 4 鼠标4 0}]
Get lock key success, processId:0, tokenId:5
Begin to process, processId:0, tokenId:5, print orderList:[{5 5 鼠标5 0}]
Get lock key success, processId:1, tokenId:6
Begin to process, processId:1, tokenId:6, print orderList:[{6 6 鼠标6 0}]
Get lock key success, processId:2, tokenId:7
Begin to process, processId:2, tokenId:7, print orderList:[{7 7 鼠标7 0}]
Get lock key success, processId:0, tokenId:8
Get lock key success, processId:1, tokenId:9
Begin to process, processId:0, tokenId:8, print orderList:[{8 8 鼠标8 0}]
Begin to process, processId:1, tokenId:9, print orderList:[{19 19 鼠标19 0}]
Get lock key success, processId:2, tokenId:0
Begin to process, processId:2, tokenId:0, print orderList:[{20 20 鼠标20 0}]
Get lock key success, processId:0, tokenId:1
Get lock key success, processId:1, tokenId:2
Begin to process, processId:0, tokenId:1, print orderList:[{11 11 鼠标11 0}]
Begin to process, processId:1, tokenId:2, print orderList:[{12 12 鼠标12 0}]
Get lock key success, processId:2, tokenId:3
Begin to process, processId:2, tokenId:3, print orderList:[{13 13 鼠标13 0}]
Get lock key success, processId:0, tokenId:4
Get lock key success, processId:1, tokenId:5
Begin to process, processId:0, tokenId:4, print orderList:[{14 14 鼠标14 0}]
Begin to process, processId:1, tokenId:5, print orderList:[{15 15 鼠标15 0}]
Get lock key success, processId:2, tokenId:6
Begin to process, processId:2, tokenId:6, print orderList:[{16 16 鼠标16 0}]
Get lock key success, processId:0, tokenId:7
Get lock key success, processId:1, tokenId:8
Begin to process, processId:0, tokenId:7, print orderList:[{17 17 鼠标17 0}]
Begin to process, processId:1, tokenId:8, print orderList:[{18 18 鼠标18 0}]
Get lock key success, processId:2, tokenId:9
Begin to process, processId:2, tokenId:9, print orderList:[]看到这里，是不是很有意思，你也可以动手实现一下~~
局限性
这个多机分片，并不是所有的异步任务都可以使用这种方式，只有特点的场景才可以：
- 一般是对DB的数据进行操作，因为DB可以很好兼容这种分片处理方式，上面的示例就是很好的说明；
- 每处理完一条DB数据，要求DB能记录数据的变更状态，对于不记录数据处理完成状态的方式，比如需要对库表中的所有用户发送Push，但是是否发送完成，DB不进行记录，只是从前往后遍历的方式处理，该分片目前不能很好支持。（如果非要支持也可以，只要你能设计一套方案，将数据切成对应的分片，保证所有机器执行的数据不重不漏，也是可以的）；
- 目前的方案只支持永动型任务，也就是任务需要一直在内存执行，不能暂停的场景，当然你也可以设计成可以支持暂停的方式，比如记录每个令牌对应的数据是否全部执行完，如果执行完了，就先暂停一段时间，然后再启动。
存在的坑
- 这里面有个坑，就是你获取分布式锁时，其实给这个锁设置了超时时间，如果超时时间过长，机器挂了，那么这个分片将会很长时间不会执行，需要等到锁自动超时；如果锁的超时时间设置过短，会导致这个分片的数据没有执行完，锁被其它线程获取，会导致同一个分片有2个线程执行，有悖我们的设计。（有没有完美的解决方案呢，其实是有的，可以加我微信，一起讨论这个问题哈~~）
后记
从限流常用方式，再到Redis分布式锁，最后是多机执行异步任务，终于把这块想写的都写完了，其实对外输出的过程，也是自己成长的过程。最近在看消息队列和设计模式，消息队列的理论部分已经写完了，前后整理了一个月，感觉头都大了，所以纯理论的内容我会尽量少些，多写一些实战方面的内容，避免自己眼高手低。设计模式的部分，我打算以实际项目为主，然后去讲解常用设计模式的实现姿势。