用 Celery 打造定时任务：Postmon 的 IBGE 数据每日自动更新实战

用 Celery 打造定时任务Postmon 的 IBGE 数据每日自动更新实战【免费下载链接】postmonUma API para consulta de informações relacionadas a endereçamento e entrega no Brasil.项目地址: https://gitcode.com/gh_mirrors/po/postmonPostmon 是一个面向巴西市场的开源 API 项目专注于提供 CEP邮政编码地址与递送信息的查询服务。为了让内置的 IBGE 官方统计信息始终保持最新Postmon 采用了 Celery 定时任务方案实现了数据的每日自动更新。这篇文章不堆砌复杂代码而是以新手友好的方式带你拆解 Postmon 是如何用 Celery 定时任务完成这项工作的。什么是 Celery 定时任务Celery 是 Python 生态中最流行的分布式任务队列框架它把需要执行的工作任务交给后台进程Worker异步处理。而定时任务则是在此基础上按照预设的时间表自动触发任务。Postmon 用到的两个关键组件组件作用Celery Worker真正执行任务的后台工作进程Celery Beat定时调度器按计划自动把任务投递给 WorkerBroker消息中间件用于传递任务Postmon 直接复用 MongoDB简单来说Celery Beat 负责到点提醒Worker 负责干活MongoDB 充当消息中转站三者协作构成一套完整的 Celery 定时任务系统。Postmon 项目速览 Postmon 是一个使用 Python 编写的开源项目技术栈非常精简Web 框架Bottle轻量级 Python Web 框架数据库MongoDB既存业务数据也当 Celery Broker定时任务Celery Celery Beat数据抓取requests 拉取外部数据源其中与 Celery 定时任务直接相关的文件包括PostmonTaskScheduler.py定时任务的核心Celery 配置与任务定义都在这里IbgeTracker.py负责抓取和整理 IBGE 数据database.pyMongoDB 数据访问层PostmonServer.pyBottle Web 服务入口README.md官方文档包含 Scheduler 的完整说明requirements.txt依赖清单含celery[mongodb]Celery 定时任务配置实战从零看懂 PostmonTaskScheduler ⚙️打开 PostmonTaskScheduler.py你会发现 Postmon 把整套定时任务浓缩在三个步骤里。第一步创建 Celery 应用并指定 Broker代码通过环境变量POSTMON_DB_HOST、POSTMON_DB_PORT等读取 MongoDB 连接信息然后创建 Celery 实例app Celery(postmon, brokerbroker_conn_string)这样 MongoDB 同时承担了业务数据库与消息 Broker 的双重角色无需额外部署 Redis 或 RabbitMQ部署成本大幅降低。第二步用 CELERYBEAT_SCHEDULE 定义定时计划Postmon 在 Celery 配置中声明了两条定时计划CELERYBEAT_SCHEDULE{ track_ibge_daily: { task: PostmonTaskScheduler.track_ibge, schedule: timedelta(days1) }, track_packs: { task: PostmonTaskScheduler.track_packs, schedule: timedelta(hours1), } }这里非常直观track_ibge_daily每 24 小时执行一次 IBGE 数据更新任务这正是每日自动更新的由来track_packs每小时执行一次包裹追踪任务第三步编写任务函数每个任务都是一个普通的 Python 函数只需加上app.task装饰器即可被 Celery 识别。比如 IBGE 更新任务app.task def track_ibge(): db Database() ibge IbgeTracker() ibge.track(db)IBGE 数据每日自动更新是怎么实现的定时任务的核心逻辑藏在 IbgeTracker.py 中它的工作流程分为三步拉取数据通过 requests 请求 IBGE 相关的 JSON 数据源州数据ufs.json与城市数据cidades.json清洗整理为每个城市生成唯一键sigla_uf_nome_cidade因为巴西存在不同州同名的城市用于数据去重写入数据库调用 database.py 中的insert_or_update_uf/insert_or_update_cidade方法以 upsert 方式更新 MongoDB 的ufs与cidades集合整个流程由track()方法串联起来。定时任务每天触发一次即可保证 API 返回的estado_info和cidade_info永远是最新数据。如何启动 Celery 定时任务最简单的方式一条命令搞定在项目根目录执行celery worker -B -A PostmonTaskScheduler -l info-B同时启动 Celery Beat 调度器-A指定 Celery 应用模块-l info输出信息级日志生产环境推荐用 Supervisord 保活对于生产环境官方建议借助 Supervisord 保持进程常驻配置示例如下[program:celeryd] commandcelery worker -B -A PostmonTaskScheduler -l info directoryPOSTMON_HOME/repositorio autostarttrue autorestarttrue startsecs10 stopwaitsecs600这样即使进程意外退出Supervisord 也会自动拉起保证 IBGE 每日自动更新任务稳定执行。使用 Docker 快速体验Postmon 提供了现成的 Dockerfile构建并运行git clone https://gitcode.com/gh_mirrors/po/postmon docker build -t postmon . docker run -d -p 80:9876 postmon容器启动时会自动拉起 MongoDB再运行 Web 服务与定时任务调度。调试 Celery 定时任务的实用技巧 ️1. 修改调度频率快速验证开发调试时可以把 PostmonTaskScheduler.py 中的timedelta(days1)临时改为timedelta(minutes1)一分钟即可验证一次任务是否正常确认无误后再改回来。2. 理解 celerybeat_schedule 文件Celery Beat 首次运行时会自动生成一个名为celerybeat_schedule的本地文件用来记录调度状态。默认生成在启动目录可通过-s参数指定其他位置celery worker -B -A PostmonTaskScheduler -l info -s /novo/caminho/celerybeat_schedule3. 关注时区配置Postmon 将时区设置为America/Sao_Paulo并启用 UTC确保每天更新一次的语义与巴西本地时间一致这也是多区域部署时最容易踩的坑。总结 通过 Postmon 这个真实案例我们看到一套完整的 Celery 定时任务只需要三块拼图Celery 应用与 Broker 配置、CELERYBEAT_SCHEDULE 定时计划、带app.task装饰器的任务函数。配合 Celery Beat 调度器一个命令即可让 IBGE 数据每日自动更新无需任何人工干预。如果你正在规划自己的 Python 定时任务方案不妨参考 Postmon 这套轻量实践——用 MongoDB 当 Broker、用 Celery 编排任务、用 Supervisord 保活简单、可靠又易于扩展。【免费下载链接】postmonUma API para consulta de informações relacionadas a endereçamento e entrega no Brasil.项目地址: https://gitcode.com/gh_mirrors/po/postmon创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考