支付宝定时任务怎么做？三层分发任务处理框架介绍

轻风博客 2023-04-24 原文

本文将从单机定时调度开始，循序渐进地带领大家了解五福定制三层分发任务处理框架。

一、背景介绍

技术同学对定时任务肯定不陌生。定时任务一般用来定时批量进行业务处理。支付宝卡包券到期提醒、删除过期失效券，五福大促批量给用户发放添福红包等场景，都是通过定时任务触发来完成的。

作者有幸参与了2023兔年五福大促的开发，主导完成了福气乐园分会场平分5000万大奖需求。通过学习并运用五福定制三层分发任务处理框架，最终平稳丝滑的完成了平分大奖需求任务。本文将从单机定时调度开始，循序渐进地带领大家了解五福定制三层分发任务处理框架。

二、定时任务分类

本文将定时任务分为单机和集群两大类别，其中单机又分为定时调度和定时调度加批处理框架，集群分为三层分发和五福定制三层分发任务处理框架。

2.1、单机任务

单机定时任务毫无疑问是在单台机器上运行的定时任务。在业务量级不大，没有进行分库分表时，往往单机定时任务即可满足业务需求。

从复杂度上来说，单机定时任务又可分为简单的定时调度和定时调度+批处理两种。

1、定时调度

在Spring中可以通过@Scheduled 来启用定时任务。触发的方式有两种，分别是：cron 表达式和 fixedRated类配置参数。常用的案例：

// cron表达式
@Scheduled(cron="0 0/30 9-17 * * ?") //按cron规则执行,朝九晚五工作时间内每半小时
@Scheduled(cron="0 0 12 ? * WED") //按cron规则执行,表示每个星期三中午12点

// fixedRated类配置
@Scheduled(fixedRate=5000) //上一次开始执行时间点后5秒再次执行；
@Scheduled(fixedDelay=3000) //上一次执行完毕时间点后3秒再次执行；
@Scheduled(initialDelay=1000, fixedDelay=2000) //第一次延迟1秒执行，然后在上一次执行完毕时间点后2秒再次执行；

定时调度往往用于业务处理流程比较简单的场景，比如定时生成简单报表，发送通知。对于复杂耗时的场景，处理效率不高，业务高峰期会积压大量待处理数据，影响业务。

2、定时调度+批处理

为了解决复杂耗时场景下定时调度效率不高的问题，可以引入批处理框架。定时调度与批处理框架相结合，可以大幅提高数据处理的效率，提升系统稳定性，保障业务稳定运行。

以Spring Batch批处理框架为例，任务处理流程如下：

Spring Batch批处理框架将任务拆分成多个Step，同时每个Step里面又分为itemReader,itemProcessor, itemWriter。通过将任务分层细化，能够让多个阶段并行处理，提高任务处理效率。批处理框架结合定时调度框架，可以在单机情况下，对大量复杂的业务进行高效的批处理。

2.2、集群任务

在分库分表大业务流量情况下，单机定时任务已无法满足业务需求了，这时就产生了集群定时任务。在支付宝技术架构下，用户数据按照eid进行分库分表，同时进行Zone维度的隔离。此时单机定时任务无法对全量数据做处理，于是支付宝便有了自己的分布式任务调度中间件Antscheduler，配合三层分发任务处理框架，就可以对大量数据进行定时批量处理。

1、三层分发

上图描述了三层分发实现定时任务处理的过程：

1.Antscheduler任务调度中间件按照配置好的规则，定时往消息中心投递消息。

2.消息中心将定时任务消息分别投递到每个Zone中的一台机器。

3.接收到消息的机器进入三层分发的第一层，即Splitor处理流程。通常是获取当前Zone的eid分片，比如00~24。

4.第一层Splitor处理完了之后，通过TR oneway的调用方式，在当前Zone对三层分发的第二层发起调用，即进入Loader处理流程。此时调用得到扩散，eid分片为00~24的情况下，会发起25次TR调用，即最多会进入当前Zone的25台机器进行Loader处理。

5.Loader通常是获取传递过来eid分片的数据。比如一台机器的Loader接收到eid为20，则该Loader从eid为20分片的DB获取100条待处理数据。

6.第二层Loader获取到待处理数据后，同样是通过TR oneway的调用方式，在当前Zone对三层分发的第三层发起调用，即进入Executor处理流程。此时调用进一步得到扩散，一个Loader获取的100条数据，此时会发起100次TR调用，即最多会进入当前Zone的100台机器进行Executor处理。

7.Executor通常是进行真正业务处理逻辑的地方。比如对每条数据做状态变更、发送eid维度消息等等。

三层分发能很好的将任务进行分层拆分扩散，充分利用机器资源，尽量做到负载均衡。但三层分发也存在一些缺陷，主要体现在以下几个方面：

1.定时调度间隔和间隔内能够处理的数据量很难完全匹配。调度间隔时间太长，机器资源没有得到有效利用，会导致处理效率低下，任务会积压；调度间隔时间太短，有可能会导致数据被重复捞取处理，为此要做额外的防重复处理逻辑。

2.无法做到平滑的任务处理。由于在Loader层获取要处理的任务数，交由Executor层执行时，并不能限制任务执行的qps，同时待处理任务数变多时，整个集群任务的qps就变得很高，对DB和其他外围系统来说，存在稳定性风险。

3.无法最大化利用集群机器资源。考虑到稳定性和高可用，设计上每个Zone是有A/B分组的，每个分组都能获取到本Zone的eid分片，所以配置Antscheduler调度规则时，通常只会选择Zone的A/B组中的一组开启任务调度。三层分发内部每个层级之间的TR调用又只能在本Zone同组内进行，A/B组之间无法进行TR调用，所以浪费了一半的机器资源。

2、五福定制三层分发

五福大促有很多业务场景都是需要通过定时任务来进行处理的，比如生肖卡提醒、AI年画提醒，福气乐园平分5000万大奖。五福大促对稳定性和可用性的要求是非常高的，为了解决三层分发处理框架缺陷带来的效率和稳定性风险，五福在三层分发基础上做了定制化改造，改造的目标主要有两点：

1）最大化利用集群机器资源。做到真正的负载均衡，同时也能够提升集群的任务处理容量。

2）平滑的任务处理。减少任务调用的尖刺，避免对DB和外部系统造成稳定性风险。

下面分别从优化目标的两点来进行阐述。

1、最大化利用集群机器资源

由于三层分发默认只会在同一个Zone的A/B组中开启一组，导致浪费了一半的机器。显而易见，要最大化利用集群机器资源，就需要让A/B组的机器都能够参与到任务处理当中。优化的步骤如下：

1.Antscheduler定时调度平台同时开启A/B分组调度。

2.增加任务配置，配置的目的是让A组的机器只处理奇数位eid、B组的机器只处理偶数位eid。

3.Splitor层根据任务配置，将本Zone全量eid进行分组，A组只处理奇数位eid,B组只处理偶数位eid。

上面三步做完以后，就能让集群所有的机器都能参与到任务处理当中，从而最大化的利用了机器资源。

以上图为例，Zone_01对应DB的eid分片为00~24，经过Splitor处理之后，Zone_01(A)组机器获得任务处理的eid都是奇数位，即01、03、...23；同理，Zone_01(B)组机器获得任务处理的eid都是偶数位，即00、02、...24。后续Loader、Executor处理都在本组内进行。通过把eid分片像A/B分组那样进行奇/偶分组，就能让所有机器都能够参与到任务处理当中。下面是eid分组的核心代码：

/**
* 根据配置中心的dataFlag过滤
* 1、默认ALL不区分
* 2、ODD 表示仅分发奇数表号
* 3、EVEN 标识仅分发偶数表号
*
* @param eidList
*/
public void filteByDataFlag(List<String> eidList) {
    String dataFlag = SchedulerConfigDrmUtil.getIndexFilterFlag();
    int strategy = -1;
    if (StringUtil.equalsIgnoreCase("ODD", dataFlag)) {
        strategy = 1;
    } else if (StringUtil.equalsIgnoreCase("EVEN", dataFlag)) {
        strategy = 0;
    }
    if (strategy == -1) {
        // ALL
        return;
    }
    // filter
    Iterator<String> it = eidList.iterator();
    while (it.hasNext()) {
        String str = it.next();
        int index = NumberUtils.toInt(str, -1);
        if (index % 2 != strategy) {
            it.remove();
        }
    }
}

通过代码可知，推送任务配置时，将A组机器的值推成“ODD”，B组机器的值推成“EVEN”，即可实现A/B组的所有机器同时执行定时任务的效果。

2、平滑的任务处理

默认的三层分发通过Loader层获取待处理的任务，然后交由Executor来执行，无法保证整个集群任务处理的量级。在待处理任务变多，或者集群机器扩缩容变化频繁的情况下，任务处理的峰值量级无法保证。同时由于各个层级之间的调用是TR oneway调用，是感知不到调用结果的，也就更难保证任务的平滑调用。为了达到任务平滑调用的目的，五福场景对Loader捞取任务数和单机任务qps做了优化调整，保证了集群任务处理效率在预期范围之内。优化的步骤如下：

1.新增任务配置。核心配置信息包括期望集群执行任务总的qps、任务调度间隔，集群参与任务机器数。

2.计算单机qps和每次DB捞取任务的数量。

3.单机执行时，根据计算好的qps来进行限流调用。

上面三步做好之后，整个集群就能够按照预期的qps进行平滑的任务处理。

为什么这三步做完了之后就能达到预期的效果呢? 重点看下任务配置的解析代码：

/**
 * 计算定时任务的相关配置
 *
 * 主要计算：
 *    scheduleSingleLimit 单机限流值
 *    scheduleLoaderCount loader捞取条数
 *
 * @param scheduleConfig
 */
public static SchedulerConfig calculateScheduleConfig(SchedulerConfig scheduleConfig) {
    final int qpsLimit = scheduleConfig.getScheduleWholeLimit();
    final int machineCounts = scheduleConfig.getScheduleMachineCounts();
    MtLogger.info(LOGGER,
        "【计算定时任务配置】-开始 任务名称:{0},机器数量:{1},任务吞吐量:{2}.", scheduleConfig.getScheduleType(),
        machineCounts, qpsLimit);

    //定时任务的调度频率是scheduleRate 秒执行一次 所以scheduleRate秒中内集群的整体吞吐量=qps限制*scheduleRate
    final int scheduleRate = scheduleConfig.getScheduleRatePerSec();
    long totalLoaderCounts = qpsLimit * TimeUnit.SECONDS.toSeconds(scheduleRate);

    //定时任务捞取的表数量为1000 所以到每个表的限制=totalLoaderCounts/1000
    long loaderCountPerTask = totalLoaderCounts / 1000;
    if (loaderCountPerTask < 1) {
        loaderCountPerTask = 1;
    }
    scheduleConfig.setScheduleLoaderCount((int) loaderCountPerTask);

    //整体限流通过单机限流实现 整体限流=单机限流*machineCounts
    final double singleQps = ((double) qpsLimit / machineCounts);
    //创建的限流需要1秒的预热
    scheduleConfig.setScheduleSingleLimit(RateLimiter.create(singleQps, 1, TimeUnit.SECONDS));

    MtLogger.info(LOGGER,
        "【计算定时任务配置】-结束 任务名称:{0},捞取条数:{1},单机限流:{2}.", scheduleConfig.getScheduleType(),
        scheduleConfig.getScheduleLoaderCount(),
        scheduleConfig.getScheduleSingleLimit().getRate());

    return scheduleConfig;

}

通过代码可知，推送的任务配置最终会生成两个重要的配置信息：

1.单个Loader捞取的任务数。集群qps和调度间隔确定了一个调度间隔内要处理的任务数，结合eid分片数量（五福是千库千表）确定每个Loader要捞取的任务数。

2.单机Executor执行任务时的qps。预期集群qps和集群机器数确定了单机执行任务时的qps，单机上通过Guava的RateLimit来达到限流的效果。如果请求超过了限制的qps，请求将会被阻塞。

下面以福气乐园平分5000万大奖的任务配置作为样例来计算：

相比通常情况下指定Loader每次捞取的任务数，五福是通过集群qps和任务调度间隔来确定Loader需要捞取的任务数。因此一个调度间隔内的任务数和集群能够执行的任务数是匹配上的，加上通过单机qps限制达到集群qps限制的效果，从而让整个定时任务做到了平滑调用。简而言之，优化后的调度逻辑，能够让定时调度任务在机器维度和时间维度都能均匀平稳的执行。

上面的截图是福气乐园平分5000万大奖真实发奖时的调用监控。通过监控可以看到，每隔5秒钟，整个集群Loader捞取的任务数是40000个。整个任务生命周期内，Executor执行任务的qps在8000上下小范围波动，没有出现大范围的波动，从而达到了平滑任务处理的预期效果。

五福定制三层分发任务处理框架，能够很好的解决任务效率和平滑调用的问题，但也存在一些使用限制。每种任务的任务配置需要人工手动推送，而其中的机器数实际上很多时候并不是固定的。推送的机器数和实际集群拥有的机器数不匹配的结果就是单机限流失准，导致集群限流不符合预期，也就导致任务间隔内捞取的任务数和能处理的任务数无法匹配上。五福场景下，机器数确定后一般不会出现大范围变动，而且五福各种业务都属于重点高保业务，机器变动后大家都能及时感知到，所以这种定制逻辑在五福场景下是可以很好的运行起来。日常的各种业务，通常集群机器数并不是固定的，存在各种扩缩容的情况，五福定制三层分发任务处理框架依旧不能保证任务的平滑调用。

三、结语

从单机到集群，再到五福定制集群定时任务，本文逐步做了一个框架设计上的原理介绍。每种定时任务都有自己的优点和缺陷，也都有自己的应用场景。在工作中，要结合当前的业务情况，选择合适的定时任务进行业务处理，避免设计上的失误导致业务受损。以五福定制三层分发任务处理框架为例，虽然日常业务中，因为机器数量不固定，依旧无法做到任务的平滑调用，但我们可以借鉴最大化利用集群机器资源这一点，同时开启A/B组的定时任务，从而实现任务调度真正的负载均衡，提高系统整体的稳定性。

作者|金盛杰(司旭)

分发任务 section data Java

有关支付宝定时任务怎么做？三层分发任务处理框架介绍的更多相关文章

ruby - 其他文件中的 Rake 任务 - 2
我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
ruby - 如何使用 RSpec::Core::RakeTask 创建 RSpec Rake 任务？ - 2
如何使用RSpec::Core::RakeTask初始化RSpecRake任务？require'rspec/core/rake_task'RSpec::Core::RakeTask.newdo|t|#whatdoIputinhere?endInitialize函数记录在http://rubydoc.info/github/rspec/rspec-core/RSpec/Core/RakeTask#initialize-instance_method没有很好的记录；它只是说:-(RakeTask)initialize(*args,&task_block)AnewinstanceofRake
Unity 热更新技术 | （三） Lua语言基本介绍及下载安装 - 2
?博客主页：https://xiaoy.blog.csdn.net?本文由呆呆敲代码的小Y原创，首发于CSDN??学习专栏推荐：Unity系统学习专栏?游戏制作专栏推荐：游戏制作?Unity实战100例专栏推荐：Unity实战100例教程?欢迎点赞?收藏⭐留言?如有错误敬请指正！?未来很长，值得我们全力奔赴更美好的生活✨------------------❤️分割线❤️-------------------------
ruby-on-rails - Rake 任务仅调用一次时执行两次 - 2
我写了一个非常简单的rake任务来尝试找到这个问题的根源。namespace:foodotaskbar::environmentdoputs'RUNNING'endend当在控制台中执行rakefoo:bar时，输出为:RUNNINGRUNNING当我执行任何rake任务时会发生这种情况。有没有人遇到过这样的事情？编辑上面的rake任务就是写在那个.rake文件中的所有内容。这是当前正在使用的Rakefile。requireFile.expand_path('../config/application',__FILE__)OurApp::Application.load_tasks这里
ruby-on-rails - 与 ActiveMerchant 一起使用的最佳支付网关是什么？ - 2
我需要使用ActiveMerchant库在我们的一个Rails应用程序中设置支付解决方案。尽管这个问题非常主观，但人们对主要网关(BrainTree、Authorize.net等)的体验如何？它必须:处理定期付款。有能力记入个人帐户。能够取消付款。有办法存储用户的付款详细信息(例如Authotize.netsCIM)。干杯最佳答案 ActiveMerchant很棒，但在过去一年左右的时间里，我在使用它时发现了一些问题。首先，虽然某些网关可能会得到“支持”——但并非所有功能都包含在内。查看功能矩阵以确保完全支持您选择的网关-http
ruby - 帮助使用 Ruby 中的 "Whenever"gem 来执行 cron 任务 - 2
我以前没有使用过cron，所以我不能确定我这样做是对的。我想要自动化的任务似乎没有运行。我在终端中执行了这些步骤:sudogeminstall每当切换到应用程序目录无论何时。(这创建了文件schedule.rb)我将此代码添加到schedule.rb:every10.minutesdorunner"User.vote",environment=>"development"endevery:hourdorunner"Digest.rss",:environment=>"development"end我将此代码添加到deploy.rb:after"deploy:symlink","depl
ruby - 在 rake 任务中运行 capybara - 2
如何在Rake任务中运行Capybara功能？例如:访问('http://google.com')谢谢! 最佳答案在任务中尝试这样的事情:require'capybara'require'capybara/dsl'Capybara.current_driver=:seleniumBrowser=Class.new{includeCapybara::DSL}page=Browser.new.pagepage.visit("http://www.google.com")puts(page.html)
ruby - Ruby 和 Ruby on Rails 中的三层架构 - 2
我是一名决定学习Ruby和RubyonRails的ASP.NETMVC开发人员。我已经有所了解并在RoR上创建了一个网站。在ASP.NETMVC上开发，我一直使用三层架构:数据层、业务层和UI(或表示)层。尝试在RubyonRails应用程序中使用这种方法，我发现没有关于它的信息(或者也许我只是找不到它？)。也许有人可以建议我如何在RubyonRails上创建或使用三层架构？附言我使用ruby1.9.3和RubyonRails3.2.3。最佳答案我建议在制作RoR应用程序时遵循RubyonRails(RoR)风格。Rails
ruby - 在 Rakefile 中动态生成 Rake 测试任务(基于现有的测试文件) - 2
我正在根据Rakefile中的现有测试文件动态生成测试任务。假设您有各种以模式命名的单元测试文件test_.rb.所以我正在做的是创建一个以“测试”命名空间内的文件名命名的任务。使用下面的代码，我可以用raketest:调用所有测试require'rake/testtask'task:default=>'test:all'namespace:testdodesc"Runalltests"Rake::TestTask.new(:all)do|t|t.test_files=FileList['test_*.rb']endFileList['test_*.rb'].eachdo|task|n
ruby-on-rails - 使用 Rspec 测试 rake 任务不接受参数 - 2
根据thispostbyStephenHagemann，我正在尝试为我的一个rake任务编写Rspec测试.lib/tasks/retry.rake:namespace:retrydotask:message,[:message_id]=>[:environment]do|t,args|TextMessage.new.resend!(args[:message_id])endendspec/tasks/retry_spec.rb:require'rails_helper'require'rake'describe'retrynamespaceraketask'dodescribe're