Micro In Action(尾声): 分布式计划任务

Micro In Action(尾声): 分布式计划任务

本文作者:Che Dan

原文链接:https://medium.com/@dche423/micro-in-action-9-cron-job-dabec09058e1

本文是《Micro In Acton》 系列的最后一篇。

我们在前面的文章中已经探讨过了利于Micro创建和使用微服务的方方面面。 今天到了最后一个话题: 计划任务。

要作好发布式计划任务不容易

基本上每个系统都需要计划任务。 它们用来执行一些后台任务, 有的是在特点时间点运行一次, 有的是定期重复执行。

如果系统里只有一个节点, 那么事情非常简单直接。我们把Web服务和后台程序都放到这个节点上, 只要这个节点是正常的, 那么后台程序就可以顺利执行。

8-1.png

但是当今的系统很少是单节点就可以搞定的, 我们需要考虑系统的鲁棒性和可扩展性。当系统的节点数是一个以后, 我们就不得不面对两个互相矛盾的问题: 重复执行单点故障

如果我们不改代码,只是往系统中填加更多完全相同的副本节点,如上图所示,必然会导致重复执行(因为每个节点中包含相同的后台程序)。要避免这些重复, 我们需要写代码去小心处理顺序执行逻辑,这很容易出错。

有些“聪明人”会调整架构。 他们把计划任务要执行的逻辑并入到Web服务中,用HTTP API 或gRPC API 把这些功能暴露出来。然后用操作系统的计划任务(例如crontab)来调用这些API,完成同样的功能。

这个架构的好处是结构简单,并且可以避免重复执行。 操作系统触发计划任务把请求发给负载均衡, 然后负载均衡会把请求转到集群中的某一个节点。

但它还是无法避免单点故障。 如果运行计划任务的那台服务器出了问题, 任务还是无法执行。

还有一些其它的替代方法, 大多无法同时解决前面提到的两个问题。

比如为后台程序单独部署一个节点, 仍存在单点故障问题。

比如为后台程序部署多个节点,仍会发生重复执行。

所以应该怎么解决这个困难吗? 说到底,我们是遇到了分布式计算中一个经典问题: 领导权选举问题

我们希望保持所有的服务节点都是完全相同(至少大部分时间如此)并无状态的。 这样就可以保证整体系统可以方便地横向扩展。但有些时间 , 我们必须在这些对等节点中选出一个主节点, 用于执行少量关键任务(例如计划任务)。

分布式选举的方案并不需要自己从头实现, 很多服务发现系统都内置提供这种能力, 例如etcd, zookeeper 等等都有此功能。

有了分布式选举,就可以保证我们的计划任务只在单一的主节点上运行(因此避免了重复执行)。而如果当前主节点因为某种原因宕机了, 其它节点会自动共同选举出新的主节点,执行后续的任务。(因为没有单点故障)

这其实就是Micro中处理计划任务所采用的方法。

Micro 中的计划任务

Micro 通过 github.com/micro/go-micro/sync 这个包提供了计划任务的功能。

不过, 与其它很多Micro功能一样,这是个“ 隐藏 ”功能。 没文档,没示例也没有测试用例可以参考。自己看代码学习,使用后果自负:joy:.

关于这个功能, 本文可能是目前全网唯一一份资料

关键组件

此功能由几个接口与类组成。

其中最重要的是 sync.Cron 接口:

// Cron is a distributed scheduler using leader election
// and distributed task runners. It uses the leader and
// task interfaces.
type Cron interface {
   Schedule(task.Schedule, task.Command) error
}

这个接口只包含一个方法 Schedule ,它接受俩参数。

第一个参数 task.Schedule 代表任务的时间计划。

// Schedule represents a time or interval at which a task should run
type Schedule struct {
   // When to start the schedule. Zero time means immediately
   Time time.Time
   // Non zero interval dictates an ongoing schedule
   Interval time.Duration
}

这个类有两个属性, 一是任务执行的起始时间点, 另一个重是执行间隔间事。

方法的第二个参数 task.Command 定义了任务要执行的命令, 包含了名称和一个函数指针(指定具体的执行代码)。

// Command to be executed
type Command struct {
   Name string
   Func func() error
}

设置计划任务

了解了上面的组件,可以很容易地创建并设置一个计划任务。

package mainimport (
   "time"   "github.com/micro/go-micro"
   "github.com/micro/go-micro/sync"
   "github.com/micro/go-micro/sync/task"
   "github.com/micro/go-micro/util/log"
)
func main() {
   // New Service
   service := micro.NewService(
      micro.Name("com.foo.cron.example"), // name the client service
   )
   // Initialise service
   service.Init()   cron := sync.NewCron()
   cron.Schedule(
      task.Schedule{Interval: 10 * time.Second},
      task.Command{Name: "foo", Func: func() error {
         log.Debug("finish command foo")
         return nil
      }},
   )
   if err := service.Run(); err != nil {
      log.Fatal(err)
   }
}

像往常一样, 先创建并初始化一个 service ,然后用 sync.NewCron() 创建一个 Cron 接口。然后用 cron.Schedule 创建设置一个简单地任务,它会每隔10秒在控制台输出一行日志。

下面把程序跑起来:

go run main.go2020-03-06 16:47:43.696426 I | Transport [http] Listening on [::]:62880
2020-03-06 16:47:43.696518 I | Broker [http] Connected to [::]:62882
2020-03-06 16:47:43.696808 I | Registry [mdns] Registering node: com.foo.cron.example-eb3a033f-4a33-438e-a467-6d2978027f2b

你可能预期每隔10秒会看到一些日志输出, 但结果是啥也没有 。不报错,也没有日志。 为啥呢?

我们知道Micro有个约定: 所有关键组件都有内置的轻量组实现。 比如有基于mDNS的注册组件, 有基于HTTP的Broker实现。 有了他们,我们可以不安装任何外部依赖,在本地完成开发。

但计划任务这个功能打破了上述约定,想使用它, 必须依赖 etcd 。看看 sync.NewCron 的源码便知道是怎么回事儿了:

func NewCron(opts ...Option) Cron {
   ...
   if options.Leader == nil {
      options.Leader = etcd.NewLeader()
   }
   ...
}

如果调用时没有提供 Leader 选项(它的类型为 leader.Leader ),那么会创建一个默认的( etcd.NewLeader )。就是这一行使得我们必须依赖etcd。如果再进一步看看 etcd.NewLeader 的代码, 就能找到根源;

func NewLeader(opts ...leader.Option) leader.Leader {
   ...
   if len(endpoints) == 0 {
      endpoints = []string{"http://127.0.0.1:2379"}
   }
   ...
}

因为我们没有etcd在 127.0.0.1:2379 地址上服务**,** 这个任务将永远卡在那儿,关键还不报错**。**

那么 leader.Leader 是什么? 我们为什么需要它呢?先看看他的定义:

// Leader provides leadership election
type Leader interface {
   // elect leader
   Elect(id string, opts ...ElectOption) (Elected, error)
   // follow the leader
   Follow() chan string
}

本文上一节提到了, 要在分布式环境中成功执行计划任务, 我们需需有领导权选举的能力,而 leader.Leader 就是用来作这个的。在sync.Cron 的内部实现中,会调用此接口的 Elect 方法来获取领导。

func (c *syncCron) Schedule(s task.Schedule, t task.Command) error {
   id := fmt.Sprintf("%s-%s", s.String(), t.String())   ...
         e, err := c.opts.Leader.Elect(id)
   ...
}

注意:我觉得 Leader 这个名字取得不好,此接口更适合命名为 Elector

所以我们需要在初始化 Cron 时提供一个 Leader 。我们可以选用etcd进行服务发现, 同时用它来作领导权选举:

import (
   ...
   "github.com/micro/go-micro/sync/leader"
   "github.com/micro/go-micro/sync/leader/etcd"
   ...
)
func main() {
   // New Service
   service := micro.NewService(
      micro.Name("com.foo.cron.example"), // name the client service
   )
   // Initialise service
   service.Init()   // get etcd node list from registry
   etcdList := service.Options().Registry.Options().Addrs
   // build leader
   lead := etcd.NewLeader(leader.Nodes(etcdList...))
   
   cron := sync.NewCron(sync.WithLeader(lead))
   cron.Schedule(
      task.Schedule{Interval: 10 * time.Second},
      task.Command{Name: "foo", Func: func() error {
         log.Info("finish command foo")
         return nil
      }},
   )   if err := service.Run(); err != nil {
      log.Fatal(err)
   }
}

接下来在启动程序时传入相关参数:

go run main.go --registry=etcd --registry_address=etcd1.foo.com:2379,etcd2.foo.com:2379,etcd3.foo.com:2379
2020-03-07 09:13:41.916437 I | Transport [http] Listening on [::]:61329
2020-03-07 09:13:41.916541 I | Broker [http] Connected to [::]:61330
2020-03-07 09:13:41.916822 I | Registry [etcd] Registering node: com.foo.cron.example-caaa1f71-2559-431d-9139-df00324250a4
2020-03-07 09:14:13.846014 I | [cron] executing command foo
2020-03-07 09:14:13.846070 I | finish command foo
...

这样, 便在Micro中启动了一个计划任务。

陷井:Micro 的计划任务依赖etcd。

如果我们再额外启动两个节点(B和C),新节点上不会有计划任务执行, 因为此时最初的节点(A)是主节点。

然后我们可以把节点A关掉,几秒钟后 B或C中的一个将自动成为新的主节点,并开始执行任务。

这样我们达到了较理想的状态: 即没有单点故障,也没有重复执行。

改变任务执行的起始时间点

有时我们希望任务在未来一个固定的时间点启动, 这就需要修改 task.ScheduleTime 属性

// start from the next New Year's Day
startPoint, _ := time.Parse("2006-01-01", "2021-01-01")
cron.Schedule(
   task.Schedule{
      Time:     startPoint,
      Interval: 10 * time.Second,
   },
   task.Command{Name: "foo", Func: func() error {
      log.Info("finish command foo")
      return nil
   }},
)

它会按你预期执行。

如果你想让任务在服务启动后的等一段时间再开始运行, 你需要这么作:

cron.Schedule(
   task.Schedule{
      Time:     time.Now().Add(time.Minute),
      Interval: 10 * time.Second,
   },
   task.Command{Name: "foo", Func: func() error {
      log.Info("finish command foo")
      return nil
   }},
)

结果 会让你失望 , 多节点中又会出现重复执行问题。问题的原因存在于实现 sync.Crontask.Schedule 的代码中。

//cron.go
func (c *syncCron) Schedule(s task.Schedule, t task.Command) error {
   id := fmt.Sprintf("%s-%s", s.String(), t.String())   ...
         e, err := c.opts.Leader.Elect(id)
   ...
}
//task.go
func (s Schedule) String() string {
   return fmt.Sprintf("%d-%d", s.Time.Unix(), s.Interval)
}

选举id依赖 Schedule.String()Schedule.String() 依赖 Schedule.Time

当我们选择以服务启动时间为基准的相对时间时, 每个节点将生成各自不同的选举id。 因为它们很难恰巧完全在相同时间点启动。

由此,每个节点自己行成了一个选举组, 自己成为自己组的主节点。

陷井: 永远别用相对时间作为任务的初次启动时间。

结语

在分布式系统中正确地实现计划任务很不容易,不过好在Micro提供了一个简单优雅的方案。尽管它缺少crontab表达式之类的高级特性,这个方案是可以满足我们大部分应用场景的。

不心点儿陷井,还是可用的。这个建议既适用于计划任务这个功能, 也适用于整个Micro框架。

至此,本系列文章就结束了。 感谢阅读。

写这个系统对我来说也是个学习的过程。 学习下来,我觉得 Micro是个相当不错的工具, 它值得学习并且真的能帮我们极大地简化分布式系统开发

缺点是有的, 不过瑕不掩瑜。 并且它也在持续升级, 我写本系列还是基于v1.18.0 , 现在已经升到v2.2.0了。 我准备去试试新版的功能。 不过, 可别着急在生产环境上使用新版本 。你要是看了我之前的文章,我肯定知道为啥这么说。

以后我还会接着下Micro相关的文章,只是不再系列的形式。 每篇文章都会独立成文,讨论某些特定的话题。

要想在我发新文章时自动得到通知, 可以在Medium 和 Twitter上 Fo 我, 帐号都是 @dche423

回头聊。感谢:pray:

— Dan Che!

我来评几句
登录后评论

已发表评论数()

相关站点

热门文章