杰瑞科技汇

Python LoopingCall如何实现定时循环任务?

Of course! In Python, the term "looping call" isn't a single built-in function, but rather a common pattern for repeatedly executing a function or block of code. The best way to implement this depends on your specific needs: synchronous (simple, blocks the program) or asynchronous (non-blocking, efficient for I/O-bound tasks).

Python LoopingCall如何实现定时循环任务?-图1
(图片来源网络,侵删)

Let's break down the most common and effective methods.


The Simple & Synchronous Way: time.sleep()

This is the most straightforward approach. You simply put the function you want to repeat inside a while loop and use time.sleep() to pause execution between calls.

Use Case: Simple tasks, background monitoring, or scripts where you don't need to do anything else while waiting.

How it works:

  1. Import the time module.
  2. Create your function to be called.
  3. In a while True loop, call the function and then time.sleep().

Example:

import time
def print_message():
    """A simple function that prints a message with the current time."""
    print(f"Task is running at: {time.ctime()}")
# --- Looping Call Implementation ---
print("Starting the synchronous looping call. Press Ctrl+C to stop.")
try:
    while True:
        print_message()
        # Wait for 5 seconds before the next call
        time.sleep(5)
except KeyboardInterrupt:
    print("\nLooping call stopped by user.")

Pros:

  • Extremely simple to understand and implement.
  • No external libraries needed.

Cons:

  • Blocking: This method pauses the entire program. If you're building a web server or a GUI application, this will freeze it completely.
  • Inflexible for complex timing (e.g., "run every 5 seconds, but adjust for the time the task itself took").

The Modern & Asynchronous Way: asyncio

For modern applications, especially those involving network requests, database calls, or GUIs, asyncio is the standard and most efficient way to handle looping calls.

Use Case: Web scraping, APIs, chat servers, any I/O-bound application where you need to handle other tasks concurrently.

How it works:

  1. Define your function as an async def function.
  2. Use asyncio.sleep() instead of time.sleep(). asyncio.sleep() is non-blocking, meaning it tells the event loop to run other tasks while it's "sleeping."
  3. Use a while True loop inside an async function.
  4. Run your main async function with asyncio.run().

Example:

import asyncio
import time
async def async_print_message():
    """An async function that prints a message."""
    print(f"Async task is running at: {time.ctime()}")
    # Simulate a non-blocking I/O operation, like a network request
    await asyncio.sleep(2) 
    print("Async task finished its work.")
async def main_loop():
    """The main async loop that runs the task repeatedly."""
    print("Starting the async looping call. Press Ctrl+C to stop.")
    try:
        while True:
            await async_print_message()
            # Wait for 5 seconds. This is non-blocking.
            await asyncio.sleep(5)
    except asyncio.CancelledError:
        print("\nAsync looping call stopped by user.")
# Run the main loop
asyncio.run(main_loop())

Pros:

  • Non-blocking: Your application can handle thousands of concurrent tasks without freezing.
  • Highly efficient for I/O-bound operations.
  • The modern standard for Python concurrency.

Cons:

  • Requires understanding the async/await paradigm.
  • All code in the asyncio loop must be "async-aware" (e.g., you can't use standard blocking libraries directly without wrapping them).

The Powerful & Flexible Way: schedule Library

If your needs are more complex than a simple fixed interval (e.g., "run every weekday at 9 AM," "run at 10:30 AM on the 15th of the month"), the schedule library is perfect.

Use Case: Cron-like scheduling, running tasks based on specific times or complex rules.

First, you need to install it:

pip install schedule

How it works:

  1. Import the schedule library.
  2. Use schedule.every().X() to define your schedule (e.g., every(5).seconds, every.day.at("09:00")).
  3. Create a while True loop that calls schedule.run_pending().

Example:

import time
import schedule
def job():
    """A job to be scheduled."""
    print(f"I'm working... Current time: {time.ctime()}")
# --- Scheduling the job ---
# Run the job every 5 seconds
schedule.every(5).seconds.do(job)
# Run the job every 10 minutes
# schedule.every(10).minutes.do(job)
# Run the job every hour at the 30th minute
# schedule.every().hour.at(":30").do(job)
# Run the job every day at a specific time
# schedule.every().day.at("10:30").do(job)
# Run the job on a specific day of the week
# schedule.every().monday.do(job)
print("Starting the scheduled looping call. Press Ctrl+C to stop.")
try:
    while True:
        # Checks all scheduled jobs and runs them if they are due
        schedule.run_pending()
        time.sleep(1) # Check every second
except KeyboardInterrupt:
    print("\nScheduled looping call stopped by user.")

Pros:

  • Very readable and intuitive syntax.
  • Extremely flexible for various scheduling needs.
  • Handles the timing logic for you.

Cons:

  • Requires an external library.
  • The main loop is still blocking (it uses time.sleep()), but you can integrate it with threading or asyncio for non-blocking behavior.

The Background Thread Way: threading

If you have a blocking function that you want to run in the background without freezing your main program, threading is the solution.

Use Case: Running a blocking task (like a file processing script) in a GUI application or a web server.

How it works:

  1. Create a function that contains your looping logic.
  2. Wrap that function in a threading.Thread.
  3. Start the thread with .start(). The thread will run in the background, and your main program will continue.

Example:

import time
import threading
def background_loop_task():
    """The task that will run in a background thread."""
    print("Background task started.")
    try:
        while True:
            print(f"Background task running at: {time.ctime()}")
            time.sleep(5)
    except KeyboardInterrupt:
        # This won't be caught by the main thread's Ctrl+C
        # You need a more robust way to stop it.
        print("Background task stopping.")
# Create and start the thread
background_thread = threading.Thread(target=background_loop_task, daemon=True)
# daemon=True means the thread will exit when the main program exits
background_thread.start()
print("Main program is running. The background thread is also running.")
print("The main program is not blocked. Let's wait for 20 seconds.")
time.sleep(20)
print("Main program finished.")

Pros:

  • Allows blocking code to run without freezing the main program.
  • Good for offloading long-running CPU-bound tasks (though for CPU-bound tasks, multiprocessing is often better).

Cons:

  • Adds complexity (thread safety, sharing data between threads).
  • Can be overkill for simple non-blocking needs where asyncio is sufficient.

Summary: Which One Should You Choose?

Method Best For Key Feature Pros Cons
time.sleep() Simple scripts, quick tasks. Simplicity. Easy, no dependencies. Blocking, inflexible.
asyncio I/O-bound apps (web, APIs). Non-blocking, efficient. Modern, scalable, handles concurrency. Requires async/await knowledge.
schedule Complex, time-based scheduling. Flexibility (cron-like). Readable, powerful scheduling. Requires external library, blocking by default.
threading Running blocking tasks in the background. Non-blocking for main thread. Good for offloading work. Adds complexity, thread safety issues.

Recommendation:

  • For most new Python projects, especially those involving network or I/O, start with asyncio.
  • If you just need a simple "run this every 5 seconds" for a script, time.sleep() is perfectly fine.
  • If your needs are based on specific times or dates, use the schedule library.
分享:
扫描分享到社交APP
上一篇
下一篇