杰瑞科技汇

Python signal 终止信号如何处理?

Of course! Let's break down Python's handling of the SIGTERM signal in detail.

Python signal 终止信号如何处理?-图1
(图片来源网络,侵删)

What is a Signal?

In Unix-like operating systems (Linux, macOS), a signal is a limited form of inter-process communication used to notify a process that an event has occurred. A process can send a signal to another process to ask it to do something, or the kernel can send a signal to a process to notify it of a system event.

Common signals include:

  • SIGINT (2): Interrupt signal. Sent when you press Ctrl+C in the terminal.
  • SIGTERM (15): Termination signal. The default signal sent by commands like kill or killall. It's a polite request for a process to terminate.
  • SIGKILL (9): Kill signal. Forcibly terminates a process. It cannot be caught or ignored, which is why it's a last resort.
  • SIGSEGV (11): Segmentation fault. Sent when a program tries to access memory it's not allowed to.

SIGTERM vs. SIGINT

Signal Name Default Action How it's Sent Common Use Case
SIGTERM Termination Terminate kill <pid> Gracefully shutting down a server or long-running script. It allows the process to clean up.
SIGINT Interrupt Terminate Ctrl+C User manually interrupting a running program. Also allows for graceful shutdown.

The key takeaway is that both SIGTERM and SIGINT can be caught and handled by a Python program, allowing it to perform cleanup operations before exiting. SIGKILL cannot.


How to Handle SIGTERM in Python

You handle signals using the signal module. The core function is signal.signal().

Python signal 终止信号如何处理?-图2
(图片来源网络,侵删)

The Basic Mechanism

  1. Import the signal module.
  2. Define a "handler" function. This is the function that will be executed when the signal is received. It should accept two arguments: the signal number and the current stack frame.
  3. Register the handler. Use signal.signal(signal.SIGTERM, your_handler_function) to tell Python to call your function when a SIGTERM is received.

Example 1: A Simple Handler

This script will print a message and then exit when it receives a SIGTERM.

# signal_handler_example.py
import signal
import time
import sys
def handle_sigterm(signum, frame):
    """
    This function is called when a SIGTERM is received.
    """
    print(f"Received signal {signum} (SIGTERM). Shutting down gracefully...")
    # --- Add your cleanup code here ---
    # e.g., close files, save state, release resources
    # ----------------------------------
    sys.exit(0)
# Register the handler for the SIGTERM signal
signal.signal(signal.SIGTERM, handle_sigterm)
print("Process is running. PID:", os.getpid())
print("To stop this process, run: kill -TERM <PID> or kill <PID>")
# Keep the script running so we can send it a signal
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    # This handles Ctrl+C (SIGINT)
    print("\nReceived SIGINT (Ctrl+C). Shutting down...")
    sys.exit(0)

How to run it:

  1. Save the code as signal_handler_example.py.
  2. Run it in your terminal: python signal_handler_example.py
  3. Note the Process ID (PID) it prints.
  4. Open a new terminal window and run the kill command: kill <PID> (replace <PID> with the actual number).
  5. You will see the "Shutting down gracefully..." message in the first terminal.

The Modern Approach: contextlib.suppress

For more complex applications, especially when using asynchronous frameworks like asyncio, the signal module can be tricky to integrate correctly. A more modern and robust approach is to use contextlib.suppress in combination with a try...finally block.

This pattern allows you to "suppress" the SystemExit exception that sys.exit() raises, ensuring your cleanup logic always runs.

Python signal 终止信号如何处理?-图3
(图片来源网络,侵删)

Example 2: Using contextlib.suppress for Robust Cleanup

This is a highly recommended pattern for building resilient services.

# robust_shutdown_example.py
import signal
import time
import sys
import os
from contextlib import suppress
# A flag to indicate that shutdown has been requested
shutdown_requested = False
def handle_shutdown(signum, frame):
    """Set a flag to indicate that shutdown is requested."""
    global shutdown_requested
    print(f"Received signal {signum} (SIGTERM). Initiating shutdown...")
    shutdown_requested = True
def cleanup():
    """This function is guaranteed to run."""
    print("Cleaning up resources...")
    # Add your actual cleanup logic here
    # e.g., close database connections, write to a log file
    time.sleep(2) # Simulate a long cleanup process
    print("Cleanup complete.")
# Register the signal handler
signal.signal(signal.SIGTERM, handle_shutdown)
signal.signal(signal.SIGINT, handle_shutdown) # Also handle Ctrl+C
print("Robust process is running. PID:", os.getpid())
print("To stop, run: kill <PID> or press Ctrl+C.")
try:
    while not shutdown_requested:
        # Main application loop
        print("Main application is running...")
        time.sleep(1)
finally:
    # This 'finally' block will ALWAYS execute, whether the loop
    # exits normally or due to an exception.
    print("Entering cleanup phase...")
    cleanup()
    print("Process has shut down successfully.")

Why this pattern is better:

  • Separation of Concerns: The signal handler's only job is to set a flag. It doesn't call sys.exit() directly, which can sometimes lead to re-entrancy issues.
  • Guaranteed Cleanup: The try...finally block ensures that cleanup() is called no matter what, even if an error occurs during the main loop.
  • Clarity: The main application loop is simple: it just checks a boolean flag.

Handling Signals in asyncio

If you are using asyncio, you should use the asyncio signal handling functions, which are designed to work correctly with the event loop.

Example 3: asyncio Signal Handling

# asyncio_signal_example.py
import asyncio
import signal
import os
async def main():
    print(f"Asyncio process is running. PID: {os.getpid()}")
    print("To stop, run: kill <PID> or press Ctrl+C.")
    # Create a flag that can be awaited
    shutdown_event = asyncio.Event()
    def handle_shutdown():
        print("Received signal. Setting shutdown event...")
        shutdown_event.set()
    # Register the signal handlers
    # loop.add_signal_handler is the correct way for asyncio
    loop = asyncio.get_running_loop()
    for sig in [signal.SIGTERM, signal.SIGINT]:
        loop.add_signal_handler(sig, handle_shutdown)
    # Main application logic
    try:
        while not shutdown_event.is_set():
            print("Asyncio application is running...")
            await asyncio.sleep(1)
    finally:
        print("Asyncio cleanup starting...")
        # Add your async cleanup logic here
        await asyncio.sleep(2)
        print("Asyncio cleanup finished.")
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nReceived KeyboardInterrupt from asyncio.run().")

Summary and Best Practices

  1. Use signal.signal() for simple, synchronous scripts.
  2. Prefer the try...finally pattern with a flag set by the signal handler. This is the most robust way to ensure cleanup logic runs in synchronous code.
  3. For asyncio, use loop.add_signal_handler(). It integrates correctly with the event loop and won't block it.
  4. Keep handlers short and fast. The signal handler should ideally just set a flag. Long-running operations in the handler can cause problems.
  5. Handle SIGINT as well. Users will often use Ctrl+C to stop a script, so handling it gracefully provides a better user experience.
分享:
扫描分享到社交APP
上一篇
下一篇