一、业务中定时任务在集群里的调度执行
1.1 什么是定时任务和集群调度
在业务里，定时任务就好比有个小闹钟，会在特定时间干某些事儿。比如每天凌晨 2 点备份数据库，或者每小时统计一下网站的访问量。而集群呢，就是好多台机器一起干活，就像一个团队。那集群调度就是安排这些机器在什么时间、怎么执行定时任务。
1.2 常见的调度方法
1.2.1 中心化调度
这种方法就像有个“老大”，所有的定时任务安排都由它说了算。“老大”会告诉每台机器什么时候该干啥。
# Python 实现简单的中心化调度示例
import schedule
import time
def job():
print("This is a scheduled job")
# 每天早上 8 点执行任务
schedule.every().day.at("08:00").do(job)
while True:
schedule.run_pending()
time.sleep(1)
这个示例里，Python 的 schedule 库就相当于“老大”，它负责安排任务在每天早上 8 点执行。
优点：好管理，所有任务安排都在一个地方，方便修改和查看。 缺点：“老大”要是出问题了，整个调度就乱套了。
1.2.2 分布式调度
分布式调度就是每台机器自己有点“小主见”，它们会根据一定规则自己决定什么时候执行任务。比如用 Redis 的分布式锁，每台机器去抢锁，抢到了就执行任务。
import redis
import time
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def job():
# 尝试获取锁
lock = r.lock('job_lock', timeout=60)
if lock.acquire():
try:
print("Executing the job")
time.sleep(10) # 模拟任务执行时间
finally:
# 释放锁
lock.release()
while True:
job()
time.sleep(60)
这个示例里，通过 Redis 的锁机制，保证同一时间只有一台机器能执行任务。
优点：可靠性高，一台机器出问题不影响其他机器执行任务。 缺点：管理起来比较复杂，要处理好锁的竞争和释放问题。
1.3 应用场景
1.3.1 数据备份
很多公司会在每天业务低谷期进行数据备份，通过集群调度可以让多台机器同时备份不同的数据，提高效率。
1.3.2 报表生成
每个月、每个季度都要生成业务报表，定时任务可以在特定时间自动从数据库取数据生成报表，集群调度能让这个过程更快速。
1.4 注意事项
- 任务依赖：有些任务之间是有先后顺序的，调度时要考虑这种依赖关系。
- 资源分配：要合理分配每台机器的任务，避免有的机器忙死，有的机器闲着。
二、Docker Swarm 运行周期性批处理作业的问题与解决办法
2.1 Docker Swarm 是啥
Docker Swarm 就像是一个“指挥官”，能把好多 Docker 容器组织起来，让它们像一个团队一样工作。周期性批处理作业就是每隔一段时间就执行一次的任务，比如每天晚上清理日志。
2.2 运行周期性批处理作业的坑
2.2.1 重复执行问题
有时候，因为网络波动或者机器故障，同一个任务可能会被执行多次，这就会浪费资源。
2.2.2 任务调度不准确
可能因为 Swarm 集群里机器的负载不均衡，导致任务不能按时执行。
2.3 解决方案
2.3.1 避免重复执行
可以用分布式锁来解决重复执行的问题。比如用 Redis 作为分布式锁的存储。
# 在 Dockerfile 中安装 Redis 客户端
FROM alpine
RUN apk add --no-cache redis
# 编写一个执行任务的脚本，使用 Redis 锁
COPY job.sh /job.sh
RUN chmod +x /job.sh
CMD ["/job.sh"]
#!/bin/sh
# job.sh
REDIS_HOST=redis_server
REDIS_PORT=6379
# 尝试获取锁
if redis-cli -h $REDIS_HOST -p $REDIS_PORT setnx job_lock 1; then
# 设置锁的过期时间，避免死锁
redis-cli -h $REDIS_HOST -p $REDIS_PORT expire job_lock 60
# 执行批处理作业
echo "Running batch job"
# 释放锁
redis-cli -h $REDIS_HOST -p $REDIS_PORT del job_lock
else
echo "Job is already running. Skipping."
fi
这个示例里，通过 Redis 的 setnx 命令尝试获取锁，获取到了才执行任务，避免了重复执行。
2.3.2 提高调度准确性
可以使用一些专门的调度工具，比如 Cronitor。它可以和 Docker Swarm 集成，更准确地调度任务。
2.4 应用场景
2.4.1 日志清理
每天晚上定时清理 Docker 容器产生的日志，避免占用过多磁盘空间。
2.4.2 数据同步
每隔一段时间从一个数据源同步数据到另一个数据源。
2.5 技术优缺点
优点：
- 可以利用 Docker 的隔离性，让不同的批处理作业互不影响。
- Docker Swarm 能自动管理容器的生命周期，方便维护。
缺点：
- 配置和管理相对复杂，需要对 Docker 和 Swarm 有一定了解。
- 网络和机器故障可能会影响任务的执行。
2.6 注意事项
- 容器资源限制：要给每个容器分配合适的资源，避免资源竞争。
- 日志监控：及时监控任务执行的日志，发现问题及时处理。
三、完整梳理避免重复执行
3.1 重复执行的危害
重复执行会浪费系统资源，比如 CPU、内存和网络带宽。还可能导致数据不一致，比如重复插入数据到数据库。
3.2 通用的避免方法
3.2.1 唯一标识
给每个任务分配一个唯一的标识，执行任务前先检查这个标识是否已经存在。
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def execute_task(task_id):
# 检查任务是否已经执行
if r.sadd('executed_tasks', task_id):
print(f"Executing task {task_id}")
# 执行任务的具体代码
else:
print(f"Task {task_id} has already been executed. Skipping.")
# 模拟执行任务
execute_task('task_123')
这个示例里，通过 Redis 的 sadd 命令检查任务标识是否已经存在，不存在就执行任务，存在就跳过。
3.2.2 时间窗口
设置一个时间窗口，在这个时间窗口内，同一个任务只执行一次。
import time
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
TIME_WINDOW = 3600 # 时间窗口为 1 小时
def execute_task(task_id):
last_execution = r.get(task_id)
if last_execution is None or (time.time() - float(last_execution)) > TIME_WINDOW:
print(f"Executing task {task_id}")
r.set(task_id, time.time())
else:
print(f"Task {task_id} was executed recently. Skipping.")
# 模拟执行任务
execute_task('task_456')
这个示例里，通过记录任务的最后执行时间，和当前时间比较，判断是否在时间窗口内，从而决定是否执行任务。
3.3 应用场景
3.3.1 数据导入
从外部数据源导入数据到系统时，避免重复导入相同的数据。
3.3.2 消息处理
处理消息队列里的消息时，避免重复处理相同的消息。
3.4 注意事项
- 唯一标识的设计：要保证唯一标识的唯一性，不能出现重复。
- 时间窗口的设置：时间窗口要根据实际业务情况合理设置，不能太短也不能太长。
四、总结
在业务里，定时任务的调度执行很重要，尤其是在集群环境中。中心化调度和分布式调度各有优缺点，要根据实际情况选择。Docker Swarm 能很好地运行周期性批处理作业，但要注意避免重复执行和提高调度准确性的问题。通过一些通用的方法，比如使用唯一标识和时间窗口，可以有效避免任务的重复执行。
在实际应用中，要根据不同的业务场景，合理选择技术和方法，同时注意资源分配、任务依赖、容器资源限制和日志监控等问题，这样才能让业务稳定、高效地运行。
评论
围绕“业务中常见定时任务在集群中如何调度执行，Docker Swarm运行周期性批处理作业的坑与解决方案，完整梳理避免重复执行”参与讨论