求各位大佬们指导批量入队列的最佳姿势

This topic created in 2987 days ago, the information mentioned may be changed or developed.

遇到个需求类似于余额宝的业务，每天按照用户的账户余额×年利率进行分红

现在的做法是每天 cron 定时处理一次将所有会员主键存入队列，然后开启一个守护进程实时监听丢给 job 处理

定时脚本里面的逻辑是从数据库取出所有会员 foreach 入队的，但是当会员数量达到百万千万越来越大的时候一次性取出的话因为内存问题不太现实，所以把定时任务从每天执行一次改成每分钟，然后每次入队一定数量的会员，但是这样没有达到理想的效果，因为每次入队一定量的数据，如果提前完成了那剩下的会员要等到下一分钟执行，而且每分钟入队最合适的数据量也需要自己估算，不能完全的实现排队执行任务

请问各位大佬通常应该用什么方式解决这个问题？

入队

会员

定时

取出

15 replies • 2018-04-14 22:43:26 +08:00

Immortal

Apr 4, 2018

“所以把定时任务从每天执行一次改成每分钟，然后每次入队一定数量的会员”
为什么不能把这一步改在脚本里面，定时启动这个脚本，在脚本里面批量处理呢。。钻牛角尖了吧

jowan

Apr 4, 2018

@Immortal 确实可以把处理的逻辑放在脚本里面但是实际处理分红逻辑的时候，会频繁操作 DB （开启事务、更新会员表，写日志等），内存会有很大的消耗，从而不得不减少每次定时处理的数量，这样每次执行的数量更少那全部执行完所有任务花费的时间更长。用队列的想法是可以开合适的多线程处理尽量减少执行时间而不用考虑内存问题

理想方式是：执行任务的总耗时 ≈ 每个任务的平均耗时 * 任务总量这样会尽快处理掉业务

Itoktsnhc

Apr 4, 2018

是否可以根据入队列的消息数量动态的注入更多的任务？

likuku

Apr 4, 2018

分布式处理机，处理节点根据自身性能开出多进程(不要让系统超载)并行处理，
假若待处理的任务太多，预计整体运行太久(超出任务最大可忍受等待时长)，
那么继续增加处理节点(能根据任务量自动增减处理节点更好)

wjpdev

Apr 4, 2018

走生产者-消费者的思路解决，自已定好队列窗口合适大小（最小值：防止消费速度过慢，最大值：防止内存占完）。两个进程，一个进程 A 实时(或间隔 1s/3s/5s)监听队列是否为空，空了就从数据库去拉下一批数据的往队列里扔。另一个守护进程 B 实时监听丢给 job 处理。

实时监控还是很占 CPU 资源的。更智能一点的，A 初始化的时候查一下当前用户总数多少，记下来。先丢 1000 个记录，第一次隔 10s 检查一下队列，如果队列空了，再丢 1000，但第二次是隔 9s 的时候检查，如果也空了，再丢 1000，隔 8s 检查。如果队列不为空，检查时间点+0.1 ..... 直到找到一个最优的检查时间点丢任务。

1000 和 0.1 是可以调节的。伪 AI .

3IOhG7M0knRu5UlC

Apr 4, 2018 via Android

千万压入队列也没多少占用，然后处理不就行了不够就多来几个

jowan

Apr 4, 2018

@GooMS 入队占用的内存不是问题问题是如何快速将 DB 中的千万条数据取出压入你在脚本中一次获取千万条的数据 foreach 一下内存肯定崩了

starmoon1994

Apr 4, 2018

前几天也遇到了这种问题需要的数据可能是 1 个也可能是 1 万个
我是做了个递归在递归里每次去库里取 100 个取够了就结束递归
类似的你可以每 N 分钟取一次数据库假设队列容量是 M 那么每次就取 M-当前未消费量

hisway

Apr 4, 2018

队列任务里去查询下一批要处理的会员，生成下个执行队列

slgz

Apr 4, 2018

@wjpdev 我之前做导出功能的时候, 就是这么处理的, 每次丢 1000. 性能还可以

jevonszmx

Apr 4, 2018

有几种办法，从多个角度解决：
1、取数据太多：导致占用内存太大的问题，可以了解一下 yield，避免一次全读内存中处理；
2、定时任务频率问题：可以使用 php pcntl 创建 daemon 守护进程，监控待处理数据。守护进程不像 crontab 那样执行频率受限（ crontab 最低是每分钟一次，跑太多时间不够，太少有点浪费），所以只要设置一个合适的频率即可，不需要算每次数量那么痛苦；

3IOhG7M0knRu5UlC

Apr 4, 2018 via Android

@jowan 按照块取同志，一次取 100 条等等。然后我觉得可以让任务发布者只发布任务携带基础信息，其他业务数据交给 work 这样就拆分了压力

3IOhG7M0knRu5UlC

Apr 4, 2018 via Android

@jowan 如果优化好数据结构只需要几百 m 内存

junan0708

Apr 14, 2018 via Android

mysql limit 不行啊？

junan0708

Apr 14, 2018 via Android

主进程循环读取，丢到队列，然后 fork 子进程消费吧，