如何在持续运行其他函数时每30秒异步写入文件？[Python]

Hey there! Let's break down what's going wrong with your current code first, then walk through a couple of solid solutions.

The Problem with Your Current Code #

Your print_a() function is an infinite loop that never exits—once it starts running, the program will never reach the func_write() call below it. That's why your file write task never gets a chance to run. Also, quick note: time() returns a float, so you'll want to convert it to a string before writing it to the file, otherwise you'll hit a TypeError.

Solution 1: Use Multithreading (Great for IO-Bound Tasks) #

Since both of your tasks spend most of their time waiting (sleeping), multithreading is a perfect fit here. We can run each function in its own thread so they execute concurrently.

Here's the adjusted code:

from time import sleep, time
import threading

def print_a():
    while True:
        print("A")
        sleep(5)

def func_write():
    while True:
        with open("/tmp/time.txt", "w", encoding="utf-8") as f:
            f.write(str(time()))  # Convert float to string to avoid TypeError
        sleep(30)

if __name__ == "__main__":
    # Create threads for each function
    thread_a = threading.Thread(target=print_a)
    thread_write = threading.Thread(target=func_write)
    
    # Start both threads
    thread_a.start()
    thread_write.start()
    
    # Keep the main thread alive to let background threads run
    thread_a.join()
    thread_write.join()

Solution 2: Use Asyncio (Modern Asynchronous Approach) #

If you wanted to use async/await, you need to restructure your code to use async-compatible sleep and run both tasks in an event loop. Here's how to do it right:

import asyncio
from time import time

async def print_a():
    while True:
        print("A")
        await asyncio.sleep(5)  # Use asyncio.sleep instead of blocking time.sleep

async def func_write():
    while True:
        with open("/tmp/time.txt", "w", encoding="utf-8") as f:
            f.write(str(time()))
        await asyncio.sleep(30)

async def main():
    # Run both tasks concurrently in the event loop
    await asyncio.gather(print_a(), func_write())

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

Key Notes #

• Both solutions let print_a run non-stop while func_write triggers every 30 seconds.
• Multithreading is straightforward if you're already comfortable with synchronous code.
• Asyncio shines if you're working with other async libraries (like aiohttp) and prefer a single-threaded concurrent workflow.

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