多机分布式执行异步任务的实现姿势

发表于 2年以前 | 总阅读数：389 次

通过多机分片执行异步任务的一种实现姿势，基于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分布式锁，最后是多机执行异步任务，终于把这块想写的都写完了，其实对外输出的过程，也是自己成长的过程。最近在看消息队列和设计模式，消息队列的理论部分已经写完了，前后整理了一个月，感觉头都大了，所以纯理论的内容我会尽量少些，多写一些实战方面的内容，避免自己眼高手低。设计模式的部分，我打算以实际项目为主，然后去讲解常用设计模式的实现姿势。