Python无限循环Worker集成HTTP健康检查端点的简便方案

针对你的SQS Worker容器健康检查需求，有两种简便的实现方式，无需引入复杂的线程通信逻辑：

方案一：Docker命令式健康检查（无需搭建Web服务） #

这种方式完全不需要在Python代码中嵌入Web服务，利用Docker的健康检查命令直接验证Worker进程的存活状态。

实现步骤： #

1. 在Worker代码中添加健康标记：在你的SQS循环中，定期更新一个健康文件的时间戳（比如每次成功迭代或处理消息时），证明进程仍在正常运行：

import os
import time

def process_sqs_messages():
    # 你的SQS消息处理逻辑
    pass

while True:
    try:
        process_sqs_messages()
        # 更新健康文件时间戳
        os.utime('/tmp/healthy', None)
    except Exception as e:
        # 处理异常，可选：如果异常严重，可以删除健康文件让检查失败
        # os.remove('/tmp/healthy')
        time.sleep(5)

2. 在Dockerfile中配置健康检查：通过HEALTHCHECK指令，检查健康文件的最后修改时间是否在指定范围内（比如5分钟内）：

FROM python:3.11-slim

# 安装依赖、复制代码等步骤...

# 健康检查：检查/tmp/healthy的修改时间是否在300秒（5分钟）内
HEALTHCHECK --interval=30s --timeout=5s --start-period=60s \
    CMD test $(($(date +%s) - $(stat -c %Y /tmp/healthy))) -lt 300 || exit 1

优势： #

• 完全不需要修改Worker的核心逻辑，仅添加一行时间戳更新代码
• 无需引入任何Web框架依赖
• 实现简单，避免线程/进程通信问题

方案二：用标准库搭建极简HTTP健康检查服务（轻量线程） #

如果需要返回更详细的健康状态（比如SQS连接可用性），可以用Python标准库的http.server在后台启动一个极简的HTTP服务，无需Flask/FastAPI这类重型框架，线程逻辑极度简化。

实现代码： #

import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import boto3

class HealthCheckHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/health-check':
            # 可选：添加自定义健康检查逻辑，比如验证SQS连接
            try:
                sqs = boto3.client('sqs')
                sqs.list_queues()  # 简单的API调用验证连接
                self.send_response(200)
                self.send_header('Content-Type', 'text/plain')
                self.end_headers()
                self.wfile.write(b'OK - SQS connection healthy')
            except Exception as e:
                self.send_response(503)
                self.end_headers()
                self.wfile.write(b'UNHEALTHY - SQS connection failed')
        else:
            self.send_response(404)
            self.end_headers()

def start_health_server():
    # 启动HTTP服务器，绑定到8000端口
    server = HTTPServer(('0.0.0.0', 8000), HealthCheckHandler)
    server.serve_forever()

# 启动健康检查线程（设置为守护线程，随主进程退出）
threading.Thread(target=start_health_server, daemon=True).start()

# 你的SQS Worker主循环
while True:
    process_sqs_messages()

Docker配置： #

在Dockerfile中添加健康检查指令，通过curl访问端点：

HEALTHCHECK --interval=30s --timeout=5s --start-period=60s \
    CMD curl -f http://localhost:8000/health-check || exit 1

优势： #

• 用标准库实现，无需额外依赖
• 线程为守护线程，主进程退出时自动终止，无需手动管理
• 可扩展自定义健康检查逻辑（如数据库连接、队列状态等）

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