Python中优雅中断循环的方法：如何实现main_loop的暂停与恢复？

Great question! Scattering pause flag checks across every line of your loop is definitely messy and error-prone—here are a few cleaner, more maintainable approaches to handle pause/resume functionality elegantly:

1. Cooperative Multitasking with Generators (Best for Single-Threaded Apps) #

Generators let you yield control back to a parent loop at strategic points, which means you can centralize your pause logic instead of duplicating it everywhere. Split your loop stages into small, yieldable steps, then let a controller loop handle the pause checks.

Example (Python):

import time
from multiprocessing import Value
import threading

# Define stages as generators that yield after key actions
def stage1():
    print("Starting Stage 1")
    time.sleep(0.5)
    yield  # Let controller check pause state
    print("Finishing Stage 1")
    yield

def stage2():
    print("Starting Stage 2")
    time.sleep(0.5)
    yield
    print("Finishing Stage 2")
    yield

def process_items(items):
    for item in items:
        print(f"Processing item {item}")
        time.sleep(0.3)
        yield  # Check pause after each item

# Centralized controller handles pause logic once
def main_controller(pause_flag):
    while True:
        # Run Stage 1 with pause checks
        for _ in stage1():
            while pause_flag.value:
                time.sleep(0.1)
        
        # Run Stage 2 with pause checks
        for _ in stage2():
            while pause_flag.value:
                time.sleep(0.1)
        
        # Process items with pause checks
        for _ in process_items([1,2,3,4]):
            while pause_flag.value:
                time.sleep(0.1)

# Example usage
pause_flag = Value('b', False)
threading.Thread(target=main_controller, args=(pause_flag,), daemon=True).start()

# Simulate pause/resume
time.sleep(3)
print("\nPausing...")
pause_flag.value = True
time.sleep(2)
print("\nResuming...")
pause_flag.value = False
time.sleep(5)

This way, you only write the pause check logic once in the controller, and each stage just signals when it's safe to pause via yield.

2. Threading with an Event Object (Great for Background Loops) #

If your loop runs in a separate thread, use a threading.Event to handle pauses cleanly. The event's wait() method blocks until the loop is resumed, so you can encapsulate the pause check in a single helper function.

Example:

import threading
import time

pause_event = threading.Event()
pause_event.set()  # Start in unpaused state

def check_pause():
    # Blocks until the event is set (resumed)
    pause_event.wait()

def main_loop():
    while True:
        # Stage 1
        print("Starting Stage 1")
        time.sleep(0.5)
        check_pause()
        print("Finishing Stage 1")
        check_pause()
        
        # Stage 2
        print("Starting Stage 2")
        time.sleep(0.5)
        check_pause()
        print("Finishing Stage 2")
        check_pause()
        
        # Process items
        for item in [1,2,3,4]:
            check_pause()
            print(f"Processing item {item}")
            time.sleep(0.3)

# Start loop in background thread
threading.Thread(target=main_loop, daemon=True).start()

# Simulate pause/resume
time.sleep(3)
print("\nPausing...")
pause_event.clear()
time.sleep(2)
print("\nResuming...")
pause_event.set()
time.sleep(5)

Here, check_pause() handles all the pause logic—you just call it at points where it's safe to halt execution. No repeated while loops!

3. Decorators to Inject Pause Checks (Retrofit Existing Code) #

If you don't want to rewrite your existing stage functions, use a decorator to automatically add pause checks after each function call.

Example:

import time
import threading

pause_flag = False

def pause_after(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        # Check pause state after function completes
        global pause_flag
        while pause_flag:
            time.sleep(0.1)
        return result
    return wrapper

# Decorate your existing stage functions
@pause_after
def do_stage1():
    print("Running Stage 1")
    time.sleep(0.5)

@pause_after
def do_stage2():
    print("Running Stage 2")
    time.sleep(0.5)

@pause_after
def process_item(item):
    print(f"Processing item {item}")
    time.sleep(0.3)

def main_loop():
    while True:
        do_stage1()
        do_stage2()
        for item in [1,2,3,4]:
            process_item(item)

# Start loop in thread
threading.Thread(target=main_loop, daemon=True).start()

# Simulate pause/resume
time.sleep(3)
print("\nPausing...")
pause_flag = True
time.sleep(2)
print("\nResuming...")
pause_flag = False
time.sleep(5)

This approach keeps your main loop clean while adding pause functionality without major rewrites.

Key Takeaways: #

• Generators are perfect for single-threaded apps where you want fine-grained control over pause points.
• Threading Events work great for background loops, letting you pause/resume without blocking the main thread.
• Decorators are ideal for retrofitting pause checks into existing codebases.

All these methods eliminate the need to copy-pause logic across every line, making your code more maintainable and elegant.

内容的提问来源于stack exchange，提问作者Víctor Martínez