Loading...
墨滴

楼仔

2021/04/25  阅读:48  主题:默认主题

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

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

欢迎大家多多点赞,更多文章,请关注微信公众号“楼仔进阶之路”,点关注,不迷路~~

楼仔

2021/04/25  阅读:48  主题:默认主题

作者介绍

楼仔