如何进行千万级别数据跑批优化
最近观看公司前辈文档，看到对大数据量跑批的优化方案，参照自己的理解和之前相关经验整理了一份优化方案~
Background
定义： 跑批通常指代的是我们应用程序在固定日期针对某一批大量数据定时进行特定的处理，在金融业务中一般跑批的场景有分户日结、账务计提、账单逾期、不良资产处理等等，它具有高连贯性特点，通常我们执行完跑批后还要对跑批数据进行进一步处理，比如发 MQ 给下游消费，数仓拉取分析等。。。
跑批最怕的就是上来就干，从不考虑涉及到第三方接口时的响应时间、大事务等问题。
Problem
针对大数据量跑批会有很多的问题，比如我们要在指定日期指定时间的大数据量去处理，还要保证处理期间尽可能的高效，在出现错误时也要进行相应的补偿措施，避免影响到其它业务等 ~
- OOM ： 查询跑批数据，未进行分片处理，随着业务纵向发展数据膨胀一旦上来，就容易导致 OOM 悲剧；
- 未对数据进行批量处理： 针对业务中间的处理未采用批量处理的思维，造成花费大量的时间，另外频繁的 IO 也是问题之一；
- 避免大事务： 直接用 @Transaction 去覆盖所有的业务是不可取的，问题定位困难不说，方法处理时间变久了；
- 下游接口的承受能力： 下游的承载能力也要在我们的考虑范围之内，比如大数量分批一直发，你是爽了，下游没有足够的能力消费就会造成灾难性的问题；
- 任务时间上的隔离： 通常大数据量跑批后面还有一些业务上的处理，对于时间和健壮性上要严格控制；
- 失败任务补偿： 分布式任务调度创建跑批任务，然后拆分子任务并发到消息队列，线程池执行任务调用远程接口，这中间的任何步骤都有可能会出问题导致任务失败；
Analyze
通过以上问题的总结，我们可以得出要完整的进行大数据量跑批任务我们的代码设计需要具备以下的几点素质：
- 健壮性： 跑批任务是要通过定时的去处理这些数据，不能因为其中一条数据出现异常从而导致整批数据无法继续进行操作，所以它必须是健壮的；
- 可靠性： 针对于异常数据我们后续可进行补偿处理，所以它必须是可靠的；
- 隔离性： 避免干扰任何其他应用程序的正常运行；
- 高性能： 通常跑批任务要处理的数据量较大，我们不能让它处理的时间过于久，这样会挤压后续的其它连贯性业务处理时间，所以我们必须考虑其性能处理；
Solution
大数据量的数据是很庞大的，如果一次性都加载到内存里面将会是灾难性的后果，因此我们要对大数据量数据进行分割处理，这是防止 OOM 必要的一环！此外，监控、异常等方法措施也要实施到位，到问题出现再补救就晚了~
1、数据库问题
使用数据库扫表问题：
遍历数据对数据库的压力是很大的，越往后速度越慢；
解决：
遍历数据库越往后查压力越大，可以设置在每次查询的时候携带上一次的极值，让你分页查找的offect永远控制在0；
2、分片广播
分片： 在生产环境中，都是采用集群部署，如果一个跑批任务只跑在一个机器上，那效率肯定很低，我们可以利用 xxl-job「分片广播」 和 「动态分片」 功能；
分布式调度幂等： 分布式任务调度只能保证准时调到一个节点上，而且通常都有失败重试的功能。所以任务幂等都是要的，一般都是通过分布式锁来实现，这里遵循简单原则使用数据库就可以了，可以通过在任务表里 insert 一条唯一的任务记录，通过唯一键来防止重复调度。
除了用唯一键，还可以在记录中增加一个状态字段，使用乐观锁来更新状态。比如开始是初始化状态，更新成正在运行的状态，更新失败说明别的节点已经在跑这个任务。当然分布式锁的实现方案有很多，比如 redis、zk 等等。
集群分布式任务调度 xxl-job： 执行器集群部署时，“分片广播” 以执行器为维度进行分片，任务路由策略选择”分片广播”情况下，一次任务调度将会广播触发对应集群中所有执行器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务；
- 分片任务场景：10个执行器的集群来处理10w条数据，每台机器只需要处理1w条数据，耗时降低10倍；
- 广播任务场景：广播执行器机器运行shell脚本、广播集群节点进行缓存更新等
// Index 是属于 Total 第几个序列（从0开始）
int shardIndex = XxlJobHelper.getShardIndex();
// Total 是总的执行器数量
int shardTotal = XxlJobHelper.getShardTotal();
3、分批获取
- 设置步长： 分派到一个 Pod 负责的数据也是庞大的，一下查出来耗时太久容易导致超时，通常我们会引入步长的概念，比如分派给 Pod 1w条数据，我们可以将它划分 10 次查出，一次查出 1k 数据，进而避免了数据库查询数据耗时太久 ~
- 空间换时间： 跑批可能会涉及到数据准备的过程，边循环跑批数据边去查找所需的数据，涉及多个 for 嵌套的循环处理时，可以采用空间换时间的思想，将数据加载到内存中进行筛选查找，但是要做好 OOM 防范措施，比如用包装类接查找出来的数据等等，毕竟内存不是无限大的！
- 深分页： 分批查询时 limit 的偏移量越大，执行时间越长。比如 limit a, b 会查询前 a + b 条数据，然后丢弃前 a 条数据，select * 会查询所有的列，也会有回表操作。我们可以使用 子查询 优化 SQL ，先查出 id 后分页，尽量用覆盖索引 来优化；
4、事务控制
- 这些操作自身是无法回滚的，这就会导致数据的不一致。可能 RPC 调用成功了，但是本地事务回滚了，可是 PRC 调用无法回滚了；
- 在事务中有远程调用，就会拉长整个事务导致本事务的数据库连接一直被占用，从而导致数据库连接池耗尽或者单个链接超时，因此要熟悉调用链路，将事务粒度控制在最小范围内；
5、充分利用服务器资源
需要充分利用服务器资源，采用多线程，MySQL的CPU在罚息期间也是低于 50%、IOPS 使用率低于 50%；
其实跑数据是 io 密集型的，不需要非得压榨服务器资源 ~
6、MQ 消费任务并行
MQ 消费消息队列的消息时要在每个节点上同时跑多个子任务才能资源利用最大化。那么就使用到线程池了，如果选择的是Kafka或者 RocketMQ，他们的客户端本来就是线程池消费的，只需要合理调整客户端参数就可以了。如果使用的是 Redis，那就需要自己创建一个线程池，然后让一个 EventLoop 线程从 Redis 队列中取任务。放入线程池中运行，因为我们已经使用 Redis 队列做缓冲，所以线程池的队列长度设为0，这里直接使用JDK提供的 SynchronousQueue。（这里以java为例）
7、动态调整并发度
跑批任务中能动态调整速度是很重要的，有 2 个地方可以进行操作：
- 任务中调用远程接口，这个速度控制其实用 Thread.sleep() 就好了。
- 控制任务并发度，就是有多少个线程同时运行任务。这个控制可以通过调整线程池的线程数来实现，但是线程池动态调整线程数比较麻烦。动态调整可以通过开源的限流组件来实现，比如 Guava 的 RateLimiter。可以在每次调用远程接口前调用限流组件来控制并发速度。
8、失败任务如何继续
一般分布式调度路径：
- 分布式 任务调度创建跑批任务；
- 拆分子任务 多线程 并发的发送到 消息队列 ；
- 线程池 执行任务调用远程接口；
在这个链条中，可能导致任务失败或者中止的原因无非下面几个。
- 服务器 Pod 因为其它业务影响重启导致任务中止；
- 任务消费过程中失败，达到最大的重试次数；
- 业务逻辑不合理或者数据膨胀导致 OOM ；
- 消费时调用远程接口超时（这个很多人专注自己的业务逻辑从而忽略第三方接口的调用）
其实解决起来也简单，因为其它因素导致失败，你需要记录下任务的进度，然后在失败的点去再次重试 ~
- 记录进度： 我们需要知道这个任务执行到哪里了，同时也要记录更新的时间，这样才知道补偿哪里，比如进行跑批捞取时，要记录我们捞取的数据区间 ~
- 任务重试： 编写一个补偿式的任务（比如FixJob），定时的去扫面处在中间态的任务，如果扫到就触发补偿机制，将这个任务改成待执行状态投入消息队列；
9、下游接口时间
跑批最怕的就是上来就干，从不考虑涉及到第三方接口时的响应时间，如果不考虑第三方接口调用时间，那么在测试时候你会发现频繁的 YGC，这是很致命的问题，属于你设计之外的事件，但也是你必须要考虑的~
解决起来也简单，在业务可以容忍的情况下，我们可以将调用接口的业务逻辑设计一个中间态，然后挂起我们的这个业务，随后用定时任务去查询我们的业务结果，在收到信息后继续我们的业务逻辑，避免它一直在内存中堆积 ~
10、线程安全
在进行跑批时，一般会采用多线程的方式进行处理，因此要考虑线程安全的问题，比如使用线程安全的容器，使用JUC包下的工具类。
11、异常 & 监控
- 异常： 要保证程序的健壮性，做好异常处理，不能因为一处报错，导致整个任务执行失败，对于异常的数据可以跳过，不影响其他数据的正常执行；
- 监控： 一般大数据量跑批是业务核心中的核心，一次异常就是很大的灾难，对业务的损伤不可预估，因此要配置相应的监控措施，在发送异常前及时察觉，进而做补偿措施；
Reference
来源：juejin.cn/post/7433315676051406888
