如何实现两个定时自调用共享变量函数的精准并行执行？

• f1每秒自调用一次：获取z，基于当前x和z更新y
• f2每5分钟精准触发（不受f1执行延迟影响）：发布y，更新x为下一个5分钟节点
• f1的下一次执行必须等待f2完成，确保使用更新后的x值
• 当前使用asyncio时f2总是被f1的当前执行延迟，需最优解决方案

原代码依赖loop.call_later调度任务，存在两个核心缺陷：

1. asyncio事件循环为单线程执行，若f1的c(z)存在耗时操作，会直接挤占f2的触发时间，无法保证f2的精准定时
2. f1的下一次调度是在上次执行结束后延迟1秒，未处理f2的执行优先级，导致f1可能在f2执行过程中继续运行，无法及时获取更新后的x

改用asyncio协程模型，结合基于绝对时间的精准延迟计算，同时用asyncio.Lock实现f1与f2的执行同步，确保f2精准触发且f1等待f2完成后再执行。

优化后代码 #

import asyncio
from datetime import datetime, timedelta


class Record:
    def __init__(self):
        self.x = None
        self.y = None
        self._lock = asyncio.Lock()  # 用于同步f1和f2的执行顺序

    async def c(self, z):
        # 模拟基于x和z计算y的逻辑，替换为实际业务代码
        return {"timestamp": self.x, "z_value": z, "computed": True}

    async def c2(self):
        # 模拟发布y的逻辑，替换为实际发布代码
        print(f"[{datetime.now()}] 发布y: {self.y}")
        return 0

    def _get_next_5min_timestamp(self):
        # 计算下一个5分钟整的时间节点（如00:00, 00:05, 00:10...）
        now = datetime.now()
        delta = timedelta(minutes=5)
        return now + (delta - (now - datetime.min) % delta)

    async def f1(self):
        while True:
            async with self._lock:
                # 获取z的逻辑，替换为实际获取方式
                z = datetime.now().timestamp()
                # 使用当前最新的x计算y
                self.y = await self.c(z)
                print(f"[{datetime.now()}] f1执行完成，当前x: {self.x}")
            # 等待1秒，锁外等待避免占用同步资源
            await asyncio.sleep(1)

    async def f2(self):
        while True:
            # 计算到下一个5分钟节点的精准延迟
            next_run_time = self._get_next_5min_timestamp()
            delay = (next_run_time - datetime.now()).total_seconds()
            await asyncio.sleep(delay)

            # 持有锁，确保f1暂停执行，等待f2完成x更新
            async with self._lock:
                print(f"[{datetime.now()}] f2触发执行")
                # 发布y
                await self.c2()
                # 更新x为下一个5分钟节点
                self.x = next_run_time
                print(f"[{datetime.now()}] f2执行完成，更新x为: {self.x}")

    async def run(self):
        # 初始化x为第一个5分钟节点
        self.x = self._get_next_5min_timestamp()
        print(f"初始化x: {self.x}")
        # 创建并行任务
        task_f1 = asyncio.create_task(self.f1())
        task_f2 = asyncio.create_task(self.f2())
        # 永久运行任务
        await asyncio.gather(task_f1, task_f2)


if __name__ == "__main__":
    asyncio.run(Record().run())

核心优化点 #

• 精准定时：f2每次计算到下一个5分钟节点的绝对延迟，通过asyncio.sleep(delay)确保精准触发——asyncio.sleep会主动让出事件循环，不会因f1的执行耗时阻塞f2调度
• 执行同步：asyncio.Lock实现互斥逻辑，f2执行时持有锁，此时f1会暂停在锁等待处，直到f2完成x更新并释放锁，确保f1后续执行使用最新的x值
• 协程化改造：将所有业务逻辑改为异步协程，避免同步代码阻塞事件循环，保证任务调度的高效性

内容的提问来源于stack exchange，提问作者TJenn