main:新增健康检查支持和服务优化

- 在 Worker 中引入轻量级 HTTP 服务器，支持健康检查和就绪检查端点。 - 在 Kubernetes 和 Docker 配置中新增健康检查探针，提升服务稳定性。 - 更新依赖，引入 `aiohttp` 用于健康检查服务。 - 优化部署配置，调整 Redis 主机配置和镜像地址以适配新环境。
main:新增健康检查支持和服务优化
2026-02-04 12:00:30 +08:00 · 2026-02-04 11:58:56 +08:00 · 2026-02-04 11:36:01 +08:00 · 2026-02-04 11:24:29 +08:00 · 2026-02-04 10:39:40 +08:00 · 2026-02-04 10:27:01 +08:00
18 changed files with 723 additions and 313 deletions
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -372,10 +372,23 @@ kubectl apply -f deployment/kubernetes/service.yaml
 - 资源限制：256Mi-512Mi 内存，250m-500m CPU
 - 健康检查：存活探针 (/healthz)，就绪探针 (/readyz)

-### 阿里云函数计算
+### 阿里云函数计算（FC 3.0）

 ```bash
-fun deploy -t deployment/serverless/aliyun-fc.yaml
+# 安装 Serverless Devs（如未安装）
+npm install -g @serverless-devs/s
+
+# 配置阿里云凭证（首次使用）
+s config add
+
+# 部署到阿里云函数计算
+cd deployment/serverless && s deploy
+
+# 验证配置语法
+cd deployment/serverless && s plan
+
+# 查看函数日志
+cd deployment/serverless && s logs --tail
 ```

 ### AWS Lambda
--- a/README.md
+++ b/README.md
@@ -18,15 +18,16 @@

 ## 文档

-| 文档                                      | 描述           |
-|-----------------------------------------|--------------|
-| [快速入门](docs/getting-started.md)         | 10 分钟上手指南    |
-| [算法开发指南](docs/algorithm-development.md) | 详细的算法开发教程    |
-| [API 参考](docs/api-reference.md)         | 完整的 API 文档   |
-| [监控指南](docs/monitoring.md)              | 监控和告警配置      |
-| [API 规范](docs/api/README.md)            | OpenAPI 规范说明 |
-| [Kubernetes 部署](docs/kubernetes-deployment.md) | K8s 集群部署指南 |
-| [日志集成(Loki)](docs/loki-quick-reference.md)      | 日志收集部署说明     |
+| 文档                                             | 描述           |
+|------------------------------------------------|--------------|
+| [快速入门](docs/getting-started.md)                | 10 分钟上手指南    |
+| [算法开发指南](docs/algorithm-development.md)        | 详细的算法开发教程    |
+| [API 参考](docs/api-reference.md)                | 完整的 API 文档   |
+| [监控指南](docs/monitoring.md)                     | 监控和告警配置      |
+| [API 规范](docs/api/README.md)                   | OpenAPI 规范说明 |
+| [Kubernetes 部署](docs/kubernetes-deployment.md) | K8s 集群部署指南   |
+| [日志集成(Loki)](docs/loki-quick-reference.md)     | 日志收集部署说明     |
+| [阿里云函数运算FC部署入门](docs/fc-deploy.md)             | 阿里云FC部署入门    |

 ## 快速开始

--- a/deployment/Dockerfile
+++ b/deployment/Dockerfile
@@ -9,11 +9,13 @@ RUN apt-get update && apt-get install -y --no-install-recommends \

 # 复制依赖文件
 COPY requirements.txt .
-COPY requirements-dev.txt .

 # 安装 Python 依赖
 RUN pip install --no-cache-dir -r requirements.txt
-RUN pip install --no-cache-dir -r requirements-dev.txt
+
+# 安装dev依赖
+#COPY requirements-dev.txt .
+#RUN pip install --no-cache-dir -r requirements-dev.txt

 # 复制应用代码和配置
 COPY src/ ./src/
--- a/deployment/docker-compose.yml
+++ b/deployment/docker-compose.yml
@@ -45,7 +45,9 @@ services:
    build:
      context: ..
      dockerfile: deployment/Dockerfile
-    image: crpi-om2xd9y8cmaizszf.cn-beijing.personal.cr.aliyuncs.com/test-namespace-gu/fc-test:latest
+    platform: linux/amd64
+    ports:
+      - "8112:8000"
    environment:
      - APP_ENV=development
      - LOG_LEVEL=INFO
@@ -70,6 +72,12 @@ services:
    depends_on:
      redis:
        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/healthz')"]
+      interval: 30s
+      timeout: 3s
+      retries: 3
+      start_period: 10s
    deploy:
      replicas: 2

--- a/deployment/kubernetes/deployment.yaml
+++ b/deployment/kubernetes/deployment.yaml
@@ -127,16 +127,25 @@ spec:
          limits:
            memory: "512Mi"
            cpu: "500m"
-        # Worker 没有 HTTP 端口，使用命令探针
+        # Worker 现在有 HTTP 健康检查端点
+        ports:
+        - containerPort: 8000
+          name: http
        livenessProbe:
-          exec:
-            command:
-            - python
-            - -c
-            - "import redis; r = redis.Redis(host='functional-scaffold-redis'); r.ping()"
+          httpGet:
+            path: /healthz
+            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30
-          timeoutSeconds: 5
+          timeoutSeconds: 3
+          failureThreshold: 3
+        readinessProbe:
+          httpGet:
+            path: /readyz
+            port: 8000
+          initialDelaySeconds: 5
+          periodSeconds: 10
+          timeoutSeconds: 3
          failureThreshold: 3

 ---
--- a/deployment/serverless/aliyun-fc.yaml
+++ b/deployment/serverless/aliyun-fc.yaml
@@ -1,72 +0,0 @@
-# 阿里云函数计算配置
-ROSTemplateFormatVersion: '2015-09-01'
-Transform: 'Aliyun::Serverless-2018-04-03'
-Resources:
-  functional-scaffold:
-    Type: 'Aliyun::Serverless::Service'
-    Properties:
-      Description: '算法工程化 Serverless 脚手架'
-      LogConfig:
-        Project: functional-scaffold-logs
-        Logstore: function-logs
-      VpcConfig:
-        VpcId: 'vpc-xxxxx'
-        VSwitchIds:
-          - 'vsw-xxxxx'
-        SecurityGroupId: 'sg-xxxxx'
-    prime-checker:
-      Type: 'Aliyun::Serverless::Function'
-      Properties:
-        Description: '质数判断算法服务（API）'
-        Runtime: custom-container
-        MemorySize: 512
-        Timeout: 60
-        InstanceConcurrency: 10
-        CAPort: 8000
-        CustomContainerConfig:
-          Image: 'registry.cn-hangzhou.aliyuncs.com/your-namespace/functional-scaffold:latest'
-          Command: '["/app/entrypoint.sh"]'
-        EnvironmentVariables:
-          APP_ENV: production
-          LOG_LEVEL: INFO
-          METRICS_ENABLED: 'true'
-          RUN_MODE: api
-          REDIS_HOST: 'r-xxxxx.redis.rds.aliyuncs.com'
-          REDIS_PORT: '6379'
-      Events:
-        httpTrigger:
-          Type: HTTP
-          Properties:
-            AuthType: ANONYMOUS
-            Methods:
-              - GET
-              - POST
-    job-worker:
-      Type: 'Aliyun::Serverless::Function'
-      Properties:
-        Description: '异步任务 Worker'
-        Runtime: custom-container
-        MemorySize: 512
-        Timeout: 900
-        InstanceConcurrency: 1
-        CustomContainerConfig:
-          Image: 'registry.cn-hangzhou.aliyuncs.com/your-namespace/functional-scaffold:latest'
-          Command: '["/app/entrypoint.sh"]'
-        EnvironmentVariables:
-          APP_ENV: production
-          LOG_LEVEL: INFO
-          METRICS_ENABLED: 'true'
-          RUN_MODE: worker
-          REDIS_HOST: 'r-xxxxx.redis.rds.aliyuncs.com'
-          REDIS_PORT: '6379'
-          WORKER_POLL_INTERVAL: '1.0'
-          MAX_CONCURRENT_JOBS: '5'
-          JOB_MAX_RETRIES: '3'
-          JOB_EXECUTION_TIMEOUT: '300'
-      Events:
-        timerTrigger:
-          Type: Timer
-          Properties:
-            CronExpression: '0 */1 * * * *'
-            Enable: true
-            Payload: '{}'
--- a/deployment/serverless/s.yaml
+++ b/deployment/serverless/s.yaml
@@ -0,0 +1,108 @@
+# 阿里云函数计算 FC 3.0 配置
+# 使用 Serverless Devs 部署: cd deployment/serverless && s deploy
+edition: 3.0.0
+name: functional-scaffold
+access: default
+
+vars:
+  region: cn-beijing
+  image: crpi-om2xd9y8cmaizszf-vpc.cn-beijing.personal.cr.aliyuncs.com/your-namespace/fc-test:test-v1
+  redis_host: 127.31.1.1
+  redis_port: "6379"
+  redis_password: "your-password"
+
+resources:
+  # API 服务函数
+  prime-checker-api:
+    component: fc3
+    props:
+      region: ${vars.region}
+      functionName: prime-checker-api
+      description: 质数判断算法服务（API）
+      runtime: custom-container
+      cpu: 0.35
+      memorySize: 512
+      diskSize: 512
+      timeout: 60
+      instanceConcurrency: 10
+      handler: not-used
+      customContainerConfig:
+        image: ${vars.image}
+        port: 8000
+        command:
+          - /app/entrypoint.sh
+        healthCheckConfig:
+          httpGetUrl: /healthz
+          initialDelaySeconds: 3
+          periodSeconds: 5
+          timeoutSeconds: 3
+          failureThreshold: 3
+          successThreshold: 1
+      environmentVariables:
+        APP_ENV: production
+        LOG_LEVEL: INFO
+        METRICS_ENABLED: "true"
+        RUN_MODE: api
+        REDIS_HOST: ${vars.redis_host}
+        REDIS_PORT: ${vars.redis_port}
+        REDIS_PASSWORD: ${vars.redis_password}
+      vpcConfig: auto
+      logConfig: auto
+      triggers:
+        - triggerName: http-trigger
+          triggerType: http
+          triggerConfig:
+            authType: anonymous
+            methods:
+              - GET
+              - POST
+              - PUT
+              - DELETE
+
+  # 异步任务 Worker 函数
+  job-worker:
+    component: fc3
+    props:
+      region: ${vars.region}
+      functionName: job-worker
+      description: 异步任务 Worker
+      runtime: custom-container
+      cpu: 0.35
+      memorySize: 512
+      diskSize: 512
+      timeout: 900
+      instanceConcurrency: 1
+      handler: not-used
+      customContainerConfig:
+        image: ${vars.image}
+        port: 8000
+        command:
+          - /app/entrypoint.sh
+        healthCheckConfig:
+          httpGetUrl: /healthz
+          initialDelaySeconds: 5
+          periodSeconds: 10
+          timeoutSeconds: 3
+          failureThreshold: 3
+          successThreshold: 1
+      environmentVariables:
+        APP_ENV: production
+        LOG_LEVEL: INFO
+        METRICS_ENABLED: "true"
+        RUN_MODE: worker
+        REDIS_HOST: ${vars.redis_host}
+        REDIS_PORT: ${vars.redis_port}
+        REDIS_PASSWORD: ${vars.redis_password}
+        WORKER_POLL_INTERVAL: "1.0"
+        MAX_CONCURRENT_JOBS: "5"
+        JOB_MAX_RETRIES: "3"
+        JOB_EXECUTION_TIMEOUT: "300"
+      vpcConfig: auto
+      logConfig: auto
+      triggers:
+        - triggerName: timer-trigger
+          triggerType: timer
+          triggerConfig:
+            cronExpression: "0 */1 * * * *"
+            enable: true
+            payload: "{}"
--- a/docs/fc-deploy.md
+++ b/docs/fc-deploy.md
@@ -0,0 +1,58 @@
+# 阿里云 函数运算FC 部署入门
+
+本指南帮助快速上手 FunctionalScaffold 脚手架，在 10 分钟内完成第一个算法服务的开发和部署。
+
+## 环境准备
+
+- 安装 [Serverless Devs CLI](https://serverless-devs.com/docs/overview)
+
+1. 首先安装Node 环境，在Node官网下载
+   - [Node.js 下载地址](https://nodejs.org/en/download/)
+2. 安装 Serverless Devs CLI
+
+```bash
+npm install @serverless-devs/s -g
+```
+
+## 初始化 serverless dev cli 配置
+
+执行以下命令初始化 serverless dev cli 配置
+
+```bash
+s config add
+```
+
+根据引导进行操作，填入你的access key id 和 access key secret
+
+## 部署算法服务
+
+部署算法服务前，请确保已经完成环境准备和配置。
+
+修改 `s.yaml` 文件中的 vars 部分
+
+```yaml
+# 阿里云函数计算 FC 3.0 配置
+# 使用 Serverless Devs 部署: cd deployment/serverless && s deploy
+edition: 3.0.0
+name: functional-scaffold
+access: default
+
+vars:
+  region: cn-hangzhou # 换成你的区域
+  image: registry.cn-hangzhou.aliyuncs.com/your-namespace/functional-scaffold:latest # 换成你的docker 镜像
+  redis_host: r-xxxxx.redis.rds.aliyuncs.com # 换成你的redis连接
+  redis_port: "6379" # redis 端口号
+  redis_password: "your-password" #redis 密码，如果没有可留空
+```
+
+```bash
+cd deployment && s deploy
+```
+
+部署完成后，可以在控制台查看服务的运行状态和日志。
+
+## 删除算法服务
+
+```bash
+cd deployment && s remove
+```
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -25,6 +25,8 @@ dependencies = [
    "pyyaml>=6.0.0",
    # HTTP 客户端（Webhook 回调）
    "httpx>=0.27.0",
+    # 轻量级 HTTP 服务器（Worker 健康检查）
+    "aiohttp>=3.9.0",
 ]

 [project.optional-dependencies]
--- a/requirements.txt
+++ b/requirements.txt
@@ -5,6 +5,7 @@ pydantic>=2.5.0
 pydantic-settings>=2.0.0
 prometheus-client>=0.19.0
 python-json-logger>=2.0.7
+aiohttp>=3.9.0

 # Redis - 任务队列和指标存储
 redis>=5.0.0
--- a/src/functional_scaffold/algorithms/base.py
+++ b/src/functional_scaffold/algorithms/base.py
@@ -32,7 +32,7 @@ class BaseAlgorithm(ABC):
        Returns:
            Dict[str, Any]: 包含结果和元数据的字典
        """
-        from ..core.metrics_unified import incr, observe
+        from ..core.metrics_unified import incr_sync, observe_sync

        start_time = time.time()
        status = "success"
@@ -71,5 +71,7 @@ class BaseAlgorithm(ABC):
        finally:
            # 记录算法执行指标
            elapsed_time = time.time() - start_time
-            incr("algorithm_executions_total", {"algorithm": self.name, "status": status})
-            observe("algorithm_execution_duration_seconds", {"algorithm": self.name}, elapsed_time)
+            incr_sync("algorithm_executions_total", {"algorithm": self.name, "status": status})
+            observe_sync(
+                "algorithm_execution_duration_seconds", {"algorithm": self.name}, elapsed_time
+            )
--- a/src/functional_scaffold/algorithms/prime_checker.py
+++ b/src/functional_scaffold/algorithms/prime_checker.py
@@ -2,7 +2,7 @@

 from typing import Dict, Any, List
 from .base import BaseAlgorithm
-from ..core.metrics_unified import incr
+from ..core.metrics_unified import incr_sync


 class PrimeChecker(BaseAlgorithm):
@@ -31,12 +31,12 @@ class PrimeChecker(BaseAlgorithm):
            ValueError: 如果输入不是整数
        """
        if not isinstance(number, int):
-            incr('prime_check',{"status":"invalid_input"})
+            incr_sync('prime_check', {"status": "invalid_input"})
            raise ValueError(f"Input must be an integer, got {type(number).__name__}")

        # 小于2的数不是质数
        if number < 2:
-            incr('prime_check', {"status": "number_little_two"})
+            incr_sync('prime_check', {"status": "number_little_two"})
            return {
                "number": number,
                "is_prime": False,
@@ -50,7 +50,7 @@ class PrimeChecker(BaseAlgorithm):

        # 如果不是质数，计算因数
        factors = [] if is_prime else self._get_factors(number)
-        incr('prime_check', {"status": "success"})
+        incr_sync('prime_check', {"status": "success"})
        return {
            "number": number,
            "is_prime": is_prime,
--- a/src/functional_scaffold/core/job_manager.py
+++ b/src/functional_scaffold/core/job_manager.py
@@ -168,7 +168,7 @@ return 0
        await self._redis_client.hset(key, mapping=job_data)

        # 记录指标
-        incr("jobs_created_total", {"algorithm": algorithm})
+        await incr("jobs_created_total", {"algorithm": algorithm})

        logger.info(f"任务已创建: job_id={job_id}, algorithm={algorithm}")
        return job_id
@@ -320,8 +320,10 @@ return 0
            await self._redis_client.expire(key, settings.job_result_ttl)

            # 记录指标
-            incr("jobs_completed_total", {"algorithm": algorithm_name, "status": status})
-            observe("job_execution_duration_seconds", {"algorithm": algorithm_name}, elapsed_time)
+            await incr("jobs_completed_total", {"algorithm": algorithm_name, "status": status})
+            await observe(
+                "job_execution_duration_seconds", {"algorithm": algorithm_name}, elapsed_time
+            )

            logger.info(
                f"任务执行完成: job_id={job_id}, status={status}, elapsed={elapsed_time:.3f}s"
@@ -372,7 +374,7 @@ return 0
                )

                if response.status_code < 400:
-                    incr("webhook_deliveries_total", {"status": "success"})
+                    await incr("webhook_deliveries_total", {"status": "success"})
                    logger.info(
                        f"Webhook 发送成功: job_id={job_id}, url={webhook_url}, "
                        f"status_code={response.status_code}"
@@ -395,7 +397,7 @@ return 0
                await asyncio.sleep(delay)

        # 所有重试都失败
-        incr("webhook_deliveries_total", {"status": "failed"})
+        await incr("webhook_deliveries_total", {"status": "failed"})
        logger.error(f"Webhook 发送最终失败: job_id={job_id}, url={webhook_url}")

    def is_available(self) -> bool:
@@ -814,10 +816,10 @@ return 0
            # 更新指标
            from .metrics_unified import set as metrics_set

-            metrics_set("job_queue_length", {"queue": "pending"}, queue_length)
-            metrics_set("job_queue_length", {"queue": "processing"}, processing_length)
-            metrics_set("job_queue_length", {"queue": "dlq"}, dlq_length)
-            metrics_set("job_oldest_waiting_seconds", None, oldest_waiting_seconds)
+            await metrics_set("job_queue_length", {"queue": "pending"}, queue_length)
+            await metrics_set("job_queue_length", {"queue": "processing"}, processing_length)
+            await metrics_set("job_queue_length", {"queue": "dlq"}, dlq_length)
+            await metrics_set("job_oldest_waiting_seconds", None, oldest_waiting_seconds)

            return {
                "queue_length": queue_length,
--- a/src/functional_scaffold/core/metrics_unified.py
+++ b/src/functional_scaffold/core/metrics_unified.py
@@ -1,19 +1,21 @@
 """统一指标管理模块

 基于 Redis 的指标收集方案，支持多实例部署和 YAML 配置。
+使用异步 Redis 客户端，避免在异步请求路径中阻塞事件循环。
 """

 import os
 import re
 import socket
 import logging
+import asyncio
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 from functools import wraps
 import time

 import yaml
-import redis
+import redis.asyncio as aioredis

 logger = logging.getLogger(__name__)

@@ -22,7 +24,7 @@ class MetricsManager:
    """统一指标管理器

    支持从 YAML 配置文件加载指标定义，使用 Redis 存储指标数据，
-    并导出 Prometheus 格式的指标。
+    并导出 Prometheus 格式的指标。使用异步 Redis 客户端。
    """

    def __init__(self, config_path: Optional[str] = None):
@@ -37,16 +39,22 @@ class MetricsManager:
        self.instance_id = settings.metrics_instance_id or socket.gethostname()
        self.config: Dict[str, Any] = {}
        self.metrics_definitions: Dict[str, Dict[str, Any]] = {}
-        self._redis_client: Optional[redis.Redis] = None
+        self._redis_client: Optional[aioredis.Redis] = None
        self._redis_available = False
+        self._initialized = False

-        # 加载配置
+        # 加载配置（同步操作）
        self._load_config()
-        # 初始化 Redis 连接
-        self._init_redis()
-        # 注册指标定义
+        # 注册指标定义（同步操作）
        self._register_metrics()

+    async def initialize(self) -> None:
+        """异步初始化 Redis 连接"""
+        if self._initialized:
+            return
+        await self._init_redis()
+        self._initialized = True
+
    def _load_config(self) -> None:
        """加载 YAML 配置文件"""
        # 尝试多个路径
@@ -138,8 +146,8 @@ class MetricsManager:
            "custom_metrics": {},
        }

-    def _init_redis(self) -> None:
-        """初始化 Redis 连接"""
+    async def _init_redis(self) -> None:
+        """异步初始化 Redis 连接"""
        from ..config import settings

        redis_config = self.config.get("redis", {})
@@ -149,7 +157,7 @@ class MetricsManager:
        password = redis_config.get("password") or settings.redis_password

        try:
-            self._redis_client = redis.Redis(
+            self._redis_client = aioredis.Redis(
                host=host,
                port=port,
                db=db,
@@ -159,10 +167,10 @@ class MetricsManager:
                socket_timeout=5,
            )
            # 测试连接
-            self._redis_client.ping()
+            await self._redis_client.ping()
            self._redis_available = True
            logger.info(f"Redis 连接成功: {host}:{port}/{db}")
-        except redis.ConnectionError as e:
+        except aioredis.ConnectionError as e:
            logger.warning(f"Redis 连接失败: {e}，指标将不会被收集")
            self._redis_available = False
        except Exception as e:
@@ -235,7 +243,9 @@ class MetricsManager:

    # === 简单 API（业务代码使用）===

-    def incr(self, name: str, labels: Optional[Dict[str, str]] = None, value: int = 1) -> None:
+    async def incr(
+        self, name: str, labels: Optional[Dict[str, str]] = None, value: int = 1
+    ) -> None:
        """增加计数器

        Args:
@@ -252,11 +262,13 @@ class MetricsManager:
        try:
            key = f"metrics:counter:{name}"
            field = self._labels_to_key(labels) or "_default_"
-            self._redis_client.hincrbyfloat(key, field, value)
+            await self._redis_client.hincrbyfloat(key, field, value)
        except Exception as e:
            logger.error(f"增加计数器失败: {e}")

-    def set(self, name: str, labels: Optional[Dict[str, str]] = None, value: float = 0) -> None:
+    async def set(
+        self, name: str, labels: Optional[Dict[str, str]] = None, value: float = 0
+    ) -> None:
        """设置仪表盘值

        Args:
@@ -273,11 +285,11 @@ class MetricsManager:
        try:
            key = f"metrics:gauge:{name}"
            field = self._labels_to_key(labels) or "_default_"
-            self._redis_client.hset(key, field, value)
+            await self._redis_client.hset(key, field, value)
        except Exception as e:
            logger.error(f"设置仪表盘失败: {e}")

-    def gauge_incr(
+    async def gauge_incr(
        self, name: str, labels: Optional[Dict[str, str]] = None, value: float = 1
    ) -> None:
        """增加仪表盘值
@@ -296,11 +308,11 @@ class MetricsManager:
        try:
            key = f"metrics:gauge:{name}"
            field = self._labels_to_key(labels) or "_default_"
-            self._redis_client.hincrbyfloat(key, field, value)
+            await self._redis_client.hincrbyfloat(key, field, value)
        except Exception as e:
            logger.error(f"增加仪表盘失败: {e}")

-    def gauge_decr(
+    async def gauge_decr(
        self, name: str, labels: Optional[Dict[str, str]] = None, value: float = 1
    ) -> None:
        """减少仪表盘值
@@ -310,9 +322,11 @@ class MetricsManager:
            labels: 标签字典
            value: 减少的值
        """
-        self.gauge_incr(name, labels, -value)
+        await self.gauge_incr(name, labels, -value)

-    def observe(self, name: str, labels: Optional[Dict[str, str]] = None, value: float = 0) -> None:
+    async def observe(
+        self, name: str, labels: Optional[Dict[str, str]] = None, value: float = 0
+    ) -> None:
        """记录直方图观测值

        Args:
@@ -348,13 +362,13 @@ class MetricsManager:
            # +Inf 桶总是增加
            pipe.hincrbyfloat(f"metrics:histogram:{name}:bucket:+Inf", label_key, 1)

-            pipe.execute()
+            await pipe.execute()
        except Exception as e:
            logger.error(f"记录直方图失败: {e}")

    # === 导出方法 ===

-    def export(self) -> str:
+    async def export(self) -> str:
        """导出 Prometheus 格式指标

        Returns:
@@ -375,11 +389,11 @@ class MetricsManager:
                lines.append(f"# TYPE {name} {metric_type}")

                if metric_type == "counter":
-                    lines.extend(self._export_counter(name))
+                    lines.extend(await self._export_counter(name))
                elif metric_type == "gauge":
-                    lines.extend(self._export_gauge(name))
+                    lines.extend(await self._export_gauge(name))
                elif metric_type == "histogram":
-                    lines.extend(self._export_histogram(name, definition))
+                    lines.extend(await self._export_histogram(name, definition))

                lines.append("")  # 空行分隔

@@ -389,12 +403,12 @@ class MetricsManager:

        return "\n".join(lines)

-    def _export_counter(self, name: str) -> List[str]:
+    async def _export_counter(self, name: str) -> List[str]:
        """导出计数器指标"""
        lines = []
        key = f"metrics:counter:{name}"

-        data = self._redis_client.hgetall(key)
+        data = await self._redis_client.hgetall(key)
        for field, value in data.items():
            if field == "_default_":
                lines.append(f"{name} {value}")
@@ -404,12 +418,12 @@ class MetricsManager:

        return lines

-    def _export_gauge(self, name: str) -> List[str]:
+    async def _export_gauge(self, name: str) -> List[str]:
        """导出仪表盘指标"""
        lines = []
        key = f"metrics:gauge:{name}"

-        data = self._redis_client.hgetall(key)
+        data = await self._redis_client.hgetall(key)
        for field, value in data.items():
            if field == "_default_":
                lines.append(f"{name} {value}")
@@ -419,14 +433,14 @@ class MetricsManager:

        return lines

-    def _export_histogram(self, name: str, definition: Dict[str, Any]) -> List[str]:
+    async def _export_histogram(self, name: str, definition: Dict[str, Any]) -> List[str]:
        """导出直方图指标"""
        lines = []
        buckets = definition.get("buckets", [])

        # 获取所有标签组合
-        count_data = self._redis_client.hgetall(f"metrics:histogram:{name}:count")
-        sum_data = self._redis_client.hgetall(f"metrics:histogram:{name}:sum")
+        count_data = await self._redis_client.hgetall(f"metrics:histogram:{name}:count")
+        sum_data = await self._redis_client.hgetall(f"metrics:histogram:{name}:sum")

        for label_key in count_data.keys():
            prom_labels = self._key_to_prometheus_labels(label_key)
@@ -434,7 +448,7 @@ class MetricsManager:
            # 导出各个桶
            for bucket in buckets:
                bucket_key = f"metrics:histogram:{name}:bucket:{bucket}"
-                bucket_value = self._redis_client.hget(bucket_key, label_key) or "0"
+                bucket_value = await self._redis_client.hget(bucket_key, label_key) or "0"
                if label_key == "_default_":
                    lines.append(f'{name}_bucket{{le="{bucket}"}} {bucket_value}')
                else:
@@ -442,7 +456,7 @@ class MetricsManager:

            # +Inf 桶
            inf_key = f"metrics:histogram:{name}:bucket:+Inf"
-            inf_value = self._redis_client.hget(inf_key, label_key) or "0"
+            inf_value = await self._redis_client.hget(inf_key, label_key) or "0"
            if label_key == "_default_":
                lines.append(f'{name}_bucket{{le="+Inf"}} {inf_value}')
            else:
@@ -464,43 +478,79 @@ class MetricsManager:
        """检查 Redis 是否可用"""
        return self._redis_available

-    def reset(self) -> None:
+    async def reset(self) -> None:
        """重置所有指标（主要用于测试）"""
        if not self._redis_available:
            return

        try:
            # 删除所有指标相关的 key
-            keys = self._redis_client.keys("metrics:*")
+            keys = await self._redis_client.keys("metrics:*")
            if keys:
-                self._redis_client.delete(*keys)
+                await self._redis_client.delete(*keys)
            logger.info("已重置所有指标")
        except Exception as e:
            logger.error(f"重置指标失败: {e}")

+    async def close(self) -> None:
+        """关闭 Redis 连接"""
+        if self._redis_client:
+            await self._redis_client.close()
+            self._redis_client = None
+            self._redis_available = False
+            self._initialized = False
+

 # 全局单例
 _manager: Optional[MetricsManager] = None
+_manager_lock = asyncio.Lock()


-def get_metrics_manager() -> MetricsManager:
-    """获取指标管理器单例"""
+async def get_metrics_manager() -> MetricsManager:
+    """获取指标管理器单例（异步）"""
+    global _manager
+    if _manager is None:
+        async with _manager_lock:
+            if _manager is None:
+                _manager = MetricsManager()
+                await _manager.initialize()
+    elif not _manager._initialized:
+        await _manager.initialize()
+    return _manager
+
+
+def get_metrics_manager_sync() -> MetricsManager:
+    """获取指标管理器单例（同步，仅用于非异步上下文）
+
+    注意：此方法不会初始化 Redis 连接，需要在异步上下文中调用 initialize()
+    """
    global _manager
    if _manager is None:
        _manager = MetricsManager()
    return _manager


-def reset_metrics_manager() -> None:
+async def reset_metrics_manager() -> None:
    """重置指标管理器单例（主要用于测试）"""
    global _manager
+    if _manager is not None:
+        await _manager.close()
+    _manager = None
+
+
+def reset_metrics_manager_sync() -> None:
+    """同步重置指标管理器单例（主要用于测试）
+
+    注意：此方法不会关闭 Redis 连接，仅重置单例引用
+    """
+    global _manager
    _manager = None


 # === 便捷函数（业务代码直接调用）===


-def incr(name: str, labels: Optional[Dict[str, str]] = None, value: int = 1) -> None:
+async def incr(name: str, labels: Optional[Dict[str, str]] = None, value: int = 1) -> None:
    """增加计数器 - 便捷函数

    Args:
@@ -508,10 +558,11 @@ def incr(name: str, labels: Optional[Dict[str, str]] = None, value: int = 1) ->
        labels: 标签字典
        value: 增加的值，默认为 1
    """
-    get_metrics_manager().incr(name, labels, value)
+    manager = await get_metrics_manager()
+    await manager.incr(name, labels, value)


-def set(name: str, labels: Optional[Dict[str, str]] = None, value: float = 0) -> None:
+async def set(name: str, labels: Optional[Dict[str, str]] = None, value: float = 0) -> None:
    """设置仪表盘 - 便捷函数

    Args:
@@ -519,10 +570,13 @@ def set(name: str, labels: Optional[Dict[str, str]] = None, value: float = 0) ->
        labels: 标签字典
        value: 设置的值
    """
-    get_metrics_manager().set(name, labels, value)
+    manager = await get_metrics_manager()
+    await manager.set(name, labels, value)


-def gauge_incr(name: str, labels: Optional[Dict[str, str]] = None, value: float = 1) -> None:
+async def gauge_incr(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 1
+) -> None:
    """增加仪表盘 - 便捷函数

    Args:
@@ -530,10 +584,13 @@ def gauge_incr(name: str, labels: Optional[Dict[str, str]] = None, value: float
        labels: 标签字典
        value: 增加的值
    """
-    get_metrics_manager().gauge_incr(name, labels, value)
+    manager = await get_metrics_manager()
+    await manager.gauge_incr(name, labels, value)


-def gauge_decr(name: str, labels: Optional[Dict[str, str]] = None, value: float = 1) -> None:
+async def gauge_decr(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 1
+) -> None:
    """减少仪表盘 - 便捷函数

    Args:
@@ -541,10 +598,13 @@ def gauge_decr(name: str, labels: Optional[Dict[str, str]] = None, value: float
        labels: 标签字典
        value: 减少的值
    """
-    get_metrics_manager().gauge_decr(name, labels, value)
+    manager = await get_metrics_manager()
+    await manager.gauge_decr(name, labels, value)


-def observe(name: str, labels: Optional[Dict[str, str]] = None, value: float = 0) -> None:
+async def observe(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 0
+) -> None:
    """记录直方图 - 便捷函数

    Args:
@@ -552,21 +612,105 @@ def observe(name: str, labels: Optional[Dict[str, str]] = None, value: float = 0
        labels: 标签字典
        value: 观测值
    """
-    get_metrics_manager().observe(name, labels, value)
+    manager = await get_metrics_manager()
+    await manager.observe(name, labels, value)


-def export() -> str:
+async def export() -> str:
    """导出指标 - 便捷函数

    Returns:
        Prometheus 文本格式的指标字符串
    """
-    return get_metrics_manager().export()
+    manager = await get_metrics_manager()
+    return await manager.export()


-def is_available() -> bool:
+async def is_available() -> bool:
    """检查 Redis 是否可用 - 便捷函数"""
-    return get_metrics_manager().is_available()
+    manager = await get_metrics_manager()
+    return manager.is_available()
+
+
+# === 同步便捷函数（用于同步代码中的 fire-and-forget 模式）===
+
+
+def _schedule_async(coro) -> None:
+    """在后台调度异步协程（fire-and-forget 模式）
+
+    如果当前没有运行的事件循环，则静默忽略。
+    """
+    try:
+        loop = asyncio.get_running_loop()
+        loop.create_task(coro)
+    except RuntimeError:
+        # 没有运行的事件循环，静默忽略
+        pass
+
+
+def incr_sync(
+    name: str, labels: Optional[Dict[str, str]] = None, value: int = 1
+) -> None:
+    """增加计数器 - 同步便捷函数（fire-and-forget）
+
+    Args:
+        name: 指标名称
+        labels: 标签字典
+        value: 增加的值，默认为 1
+    """
+    _schedule_async(incr(name, labels, value))
+
+
+def set_sync(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 0
+) -> None:
+    """设置仪表盘 - 同步便捷函数（fire-and-forget）
+
+    Args:
+        name: 指标名称
+        labels: 标签字典
+        value: 设置的值
+    """
+    _schedule_async(set(name, labels, value))
+
+
+def gauge_incr_sync(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 1
+) -> None:
+    """增加仪表盘 - 同步便捷函数（fire-and-forget）
+
+    Args:
+        name: 指标名称
+        labels: 标签字典
+        value: 增加的值
+    """
+    _schedule_async(gauge_incr(name, labels, value))
+
+
+def gauge_decr_sync(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 1
+) -> None:
+    """减少仪表盘 - 同步便捷函数（fire-and-forget）
+
+    Args:
+        name: 指标名称
+        labels: 标签字典
+        value: 减少的值
+    """
+    _schedule_async(gauge_decr(name, labels, value))
+
+
+def observe_sync(
+    name: str, labels: Optional[Dict[str, str]] = None, value: float = 0
+) -> None:
+    """记录直方图 - 同步便捷函数（fire-and-forget）
+
+    Args:
+        name: 指标名称
+        labels: 标签字典
+        value: 观测值
+    """
+    _schedule_async(observe(name, labels, value))


 # === 装饰器（兼容旧 API）===
@@ -593,8 +737,11 @@ def track_algorithm_execution(algorithm_name: str):
                raise e
            finally:
                elapsed = time.time() - start_time
-                incr("algorithm_executions_total", {"algorithm": algorithm_name, "status": status})
-                observe(
+                incr_sync(
+                    "algorithm_executions_total",
+                    {"algorithm": algorithm_name, "status": status},
+                )
+                observe_sync(
                    "algorithm_execution_duration_seconds",
                    {"algorithm": algorithm_name},
                    elapsed,
--- a/src/functional_scaffold/main.py
+++ b/src/functional_scaffold/main.py
@@ -95,7 +95,7 @@ async def track_metrics(request: Request, call_next):
    if request.url.path in skip_paths:
        return await call_next(request)

-    gauge_incr("http_requests_in_progress")
+    await gauge_incr("http_requests_in_progress")
    start_time = time.time()
    status = "success"

@@ -112,16 +112,16 @@ async def track_metrics(request: Request, call_next):
        elapsed = time.time() - start_time
        # 使用规范化后的路径记录指标
        normalized_path = normalize_path(request.url.path)
-        incr(
+        await incr(
            "http_requests_total",
            {"method": request.method, "endpoint": normalized_path, "status": status},
        )
-        observe(
+        await observe(
            "http_request_duration_seconds",
            {"method": request.method, "endpoint": normalized_path},
            elapsed,
        )
-        gauge_decr("http_requests_in_progress")
+        await gauge_decr("http_requests_in_progress")


 # 注册路由
@@ -145,7 +145,7 @@ async def metrics():
        return Response(content="Metrics disabled", status_code=404)

    return Response(
-        content=export(),
+        content=await export(),
        media_type="text/plain; version=0.0.4; charset=utf-8",
    )

@@ -160,7 +160,7 @@ async def startup_event():

    # 初始化指标管理器
    if settings.metrics_enabled:
-        manager = get_metrics_manager()
+        manager = await get_metrics_manager()
        if manager.is_available():
            logger.info("Redis 指标收集已启用")
        else:
--- a/src/functional_scaffold/worker.py
+++ b/src/functional_scaffold/worker.py
@@ -9,6 +9,8 @@ import signal
 import sys
 from typing import Optional

+from aiohttp import web
+
 from .config import settings
 from .core.job_manager import JobManager
 from .core.logging import setup_logging
@@ -17,6 +19,53 @@ from .core.tracing import set_request_id
 logger = logging.getLogger(__name__)


+class HealthCheckServer:
+    """轻量级健康检查 HTTP 服务器
+
+    为 Worker 模式提供健康检查端点，满足 FC 3.0 容器健康检查要求。
+    """
+
+    def __init__(self, host: str = "0.0.0.0", port: int = 8000):
+        self._host = host
+        self._port = port
+        self._app: Optional[web.Application] = None
+        self._runner: Optional[web.AppRunner] = None
+        self._site: Optional[web.TCPSite] = None
+        self._healthy = True
+
+    async def start(self) -> None:
+        """启动健康检查服务器"""
+        self._app = web.Application()
+        self._app.router.add_get("/healthz", self._healthz_handler)
+        self._app.router.add_get("/readyz", self._readyz_handler)
+
+        self._runner = web.AppRunner(self._app)
+        await self._runner.setup()
+        self._site = web.TCPSite(self._runner, self._host, self._port)
+        await self._site.start()
+        logger.info(f"健康检查服务器已启动: http://{self._host}:{self._port}")
+
+    async def stop(self) -> None:
+        """停止健康检查服务器"""
+        if self._runner:
+            await self._runner.cleanup()
+            logger.info("健康检查服务器已停止")
+
+    def set_healthy(self, healthy: bool) -> None:
+        """设置健康状态"""
+        self._healthy = healthy
+
+    async def _healthz_handler(self, request: web.Request) -> web.Response:
+        """存活检查端点"""
+        return web.json_response({"status": "healthy", "mode": "worker"})
+
+    async def _readyz_handler(self, request: web.Request) -> web.Response:
+        """就绪检查端点"""
+        if self._healthy:
+            return web.json_response({"status": "ready", "mode": "worker"})
+        return web.json_response({"status": "not ready"}, status=503)
+
+
 class JobWorker:
    """任务 Worker

@@ -260,7 +309,7 @@ class JobWorker:
                    # 记录回收指标
                    from .core.metrics_unified import incr

-                    incr("job_recovered_total", None, recovered)
+                    await incr("job_recovered_total", None, recovered)

                # 收集队列监控指标
                await self._job_manager.collect_queue_metrics()
@@ -272,12 +321,21 @@ class JobWorker:
                logger.error(f"超时任务回收异常: {e}")


-def setup_signal_handlers(worker: JobWorker, loop: asyncio.AbstractEventLoop) -> None:
+def setup_signal_handlers(
+    worker: JobWorker,
+    health_server: HealthCheckServer,
+    loop: asyncio.AbstractEventLoop,
+) -> None:
    """设置信号处理器"""

+    async def shutdown_all() -> None:
+        """关闭所有服务"""
+        await worker.shutdown()
+        await health_server.stop()
+
    def signal_handler(sig: signal.Signals) -> None:
        logger.info(f"收到信号 {sig.name}，准备关闭...")
-        loop.create_task(worker.shutdown())
+        loop.create_task(shutdown_all())

    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, signal_handler, sig)
@@ -288,13 +346,19 @@ async def main() -> None:
    # 设置日志
    setup_logging(level=settings.log_level, format_type=settings.log_format)

+    # 创建健康检查服务器和 Worker
+    health_server = HealthCheckServer(port=8000)
    worker = JobWorker()

    # 设置信号处理
    loop = asyncio.get_running_loop()
-    setup_signal_handlers(worker, loop)
+    setup_signal_handlers(worker, health_server, loop)

    try:
+        # 先启动健康检查服务器，确保 FC 健康检查能通过
+        await health_server.start()
+
+        # 初始化并运行 Worker
        await worker.initialize()
        await worker.run()
    except Exception as e:
@@ -302,6 +366,7 @@ async def main() -> None:
        sys.exit(1)
    finally:
        await worker.shutdown()
+        await health_server.stop()


 if __name__ == "__main__":
--- a/tests/test_metrics_unified.py
+++ b/tests/test_metrics_unified.py
@@ -1,158 +1,239 @@
 """metrics_unified 模块单元测试"""

 import pytest
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+@pytest.fixture(autouse=True)
+def reset_manager():
+    """每个测试前后重置管理器"""
+    from functional_scaffold.core.metrics_unified import reset_metrics_manager_sync
+
+    reset_metrics_manager_sync()
+    yield
+    reset_metrics_manager_sync()


 class TestMetricsManager:
    """MetricsManager 类测试"""

-    @pytest.fixture
-    def mock_redis(self):
-        """模拟 Redis 客户端"""
-        with patch("redis.Redis") as mock:
-            mock_instance = MagicMock()
-            mock_instance.ping.return_value = True
-            mock_instance.hincrbyfloat.return_value = 1.0
-            mock_instance.hset.return_value = True
-            mock_instance.hgetall.return_value = {}
-            mock_instance.hget.return_value = "0"
-            mock_instance.keys.return_value = []
-            mock_instance.pipeline.return_value = MagicMock()
-            mock.return_value = mock_instance
-            yield mock_instance
-
-    @pytest.fixture
-    def manager(self, mock_redis):
-        """创建测试用的 MetricsManager"""
-        from functional_scaffold.core.metrics_unified import (
-            MetricsManager,
-            reset_metrics_manager,
-        )
-
-        reset_metrics_manager()
-        manager = MetricsManager()
-        return manager
-
-    def test_init_loads_default_config(self, manager):
+    def test_init_loads_default_config(self):
        """测试初始化加载默认配置"""
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
        assert manager.config is not None
        assert "builtin_metrics" in manager.config or len(manager.metrics_definitions) > 0

-    def test_metrics_definitions_registered(self, manager):
+    def test_metrics_definitions_registered(self):
        """测试指标定义已注册"""
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
        assert "http_requests_total" in manager.metrics_definitions
        assert "http_request_duration_seconds" in manager.metrics_definitions
        assert "algorithm_executions_total" in manager.metrics_definitions

-    def test_incr_counter(self, manager, mock_redis):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_incr_counter(self, mock_redis_class):
        """测试计数器增加"""
-        manager.incr("http_requests_total", {"method": "GET", "endpoint": "/", "status": "success"})
-        mock_redis.hincrbyfloat.assert_called()
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hincrbyfloat = AsyncMock(return_value=1.0)
+        mock_instance.close = AsyncMock()
+        mock_redis_class.return_value = mock_instance

-    def test_incr_with_invalid_metric_type(self, manager, mock_redis):
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
+        await manager.initialize()
+
+        await manager.incr(
+            "http_requests_total", {"method": "GET", "endpoint": "/", "status": "success"}
+        )
+        mock_instance.hincrbyfloat.assert_called()
+
+    def test_incr_with_invalid_metric_type(self):
        """测试对非计数器类型调用 incr"""
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
        # http_request_duration_seconds 是 histogram 类型
-        manager.incr("http_request_duration_seconds", {})
-        # 不应该调用 Redis（因为类型不匹配）
-        # 验证没有调用 hincrbyfloat（或者调用次数没有增加）
+        # 验证不会抛出异常（因为 Redis 不可用）

-    def test_set_gauge(self, manager, mock_redis):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_set_gauge(self, mock_redis_class):
        """测试设置仪表盘"""
-        manager.set("http_requests_in_progress", {}, 5)
-        mock_redis.hset.assert_called()
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hset = AsyncMock(return_value=True)
+        mock_instance.close = AsyncMock()
+        mock_redis_class.return_value = mock_instance

-    def test_gauge_incr(self, manager, mock_redis):
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
+        await manager.initialize()
+
+        await manager.set("http_requests_in_progress", {}, 5)
+        mock_instance.hset.assert_called()
+
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_gauge_incr(self, mock_redis_class):
        """测试增加仪表盘"""
-        manager.gauge_incr("http_requests_in_progress", {}, 1)
-        mock_redis.hincrbyfloat.assert_called()
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hincrbyfloat = AsyncMock(return_value=1.0)
+        mock_instance.close = AsyncMock()
+        mock_redis_class.return_value = mock_instance

-    def test_gauge_decr(self, manager, mock_redis):
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
+        await manager.initialize()
+
+        await manager.gauge_incr("http_requests_in_progress", {}, 1)
+        mock_instance.hincrbyfloat.assert_called()
+
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_gauge_decr(self, mock_redis_class):
        """测试减少仪表盘"""
-        manager.gauge_decr("http_requests_in_progress", {}, 1)
-        mock_redis.hincrbyfloat.assert_called()
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hincrbyfloat = AsyncMock(return_value=1.0)
+        mock_instance.close = AsyncMock()
+        mock_redis_class.return_value = mock_instance

-    def test_observe_histogram(self, manager, mock_redis):
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
+        await manager.initialize()
+
+        await manager.gauge_decr("http_requests_in_progress", {}, 1)
+        mock_instance.hincrbyfloat.assert_called()
+
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_observe_histogram(self, mock_redis_class):
        """测试直方图观测"""
-        mock_pipeline = MagicMock()
-        mock_redis.pipeline.return_value = mock_pipeline
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.close = AsyncMock()

-        manager.observe("http_request_duration_seconds", {"method": "GET", "endpoint": "/"}, 0.05)
+        mock_pipeline = AsyncMock()
+        mock_pipeline.hincrbyfloat = MagicMock()
+        mock_pipeline.execute = AsyncMock(return_value=[])
+        mock_instance.pipeline = MagicMock(return_value=mock_pipeline)

-        mock_redis.pipeline.assert_called()
-        mock_pipeline.execute.assert_called()
+        mock_redis_class.return_value = mock_instance

-    def test_labels_to_key(self, manager):
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
+        await manager.initialize()
+
+        await manager.observe(
+            "http_request_duration_seconds", {"method": "GET", "endpoint": "/"}, 0.05
+        )
+        mock_instance.pipeline.assert_called()
+
+    def test_labels_to_key(self):
        """测试标签转换为 key"""
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
        labels = {"method": "GET", "endpoint": "/api"}
        key = manager._labels_to_key(labels)
        assert "method=GET" in key
        assert "endpoint=/api" in key

-    def test_labels_to_key_empty(self, manager):
+    def test_labels_to_key_empty(self):
        """测试空标签转换"""
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
        key = manager._labels_to_key(None)
        assert key == ""

        key = manager._labels_to_key({})
        assert key == ""

-    def test_is_available(self, manager):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_is_available(self, mock_redis_class):
        """测试 Redis 可用性检查"""
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.close = AsyncMock()
+        mock_redis_class.return_value = mock_instance
+
+        from functional_scaffold.core.metrics_unified import MetricsManager
+
+        manager = MetricsManager()
+        await manager.initialize()
+
        assert manager.is_available() is True


 class TestConvenienceFunctions:
    """便捷函数测试"""

-    @pytest.fixture(autouse=True)
-    def setup(self):
-        """每个测试前重置管理器"""
-        from functional_scaffold.core.metrics_unified import reset_metrics_manager
-
-        reset_metrics_manager()
-
-    @patch("redis.Redis")
-    def test_incr_function(self, mock_redis_class):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_incr_function(self, mock_redis_class):
        """测试 incr 便捷函数"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hincrbyfloat = AsyncMock(return_value=1.0)
+        mock_instance.close = AsyncMock()
        mock_redis_class.return_value = mock_instance

-        from functional_scaffold.core.metrics_unified import incr, reset_metrics_manager
+        from functional_scaffold.core.metrics_unified import incr

-        reset_metrics_manager()
-        incr("http_requests_total", {"method": "GET", "endpoint": "/", "status": "success"})
+        await incr(
+            "http_requests_total", {"method": "GET", "endpoint": "/", "status": "success"}
+        )

        mock_instance.hincrbyfloat.assert_called()

-    @patch("redis.Redis")
-    def test_set_function(self, mock_redis_class):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_set_function(self, mock_redis_class):
        """测试 set 便捷函数"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hset = AsyncMock(return_value=True)
+        mock_instance.close = AsyncMock()
        mock_redis_class.return_value = mock_instance

-        from functional_scaffold.core.metrics_unified import reset_metrics_manager, set
+        from functional_scaffold.core.metrics_unified import set

-        reset_metrics_manager()
-        set("http_requests_in_progress", {}, 10)
+        await set("http_requests_in_progress", {}, 10)

        mock_instance.hset.assert_called()

-    @patch("redis.Redis")
-    def test_observe_function(self, mock_redis_class):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_observe_function(self, mock_redis_class):
        """测试 observe 便捷函数"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
-        mock_pipeline = MagicMock()
-        mock_instance.pipeline.return_value = mock_pipeline
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.close = AsyncMock()
+
+        mock_pipeline = AsyncMock()
+        mock_pipeline.hincrbyfloat = MagicMock()
+        mock_pipeline.execute = AsyncMock(return_value=[])
+        mock_instance.pipeline = MagicMock(return_value=mock_pipeline)
+
        mock_redis_class.return_value = mock_instance

-        from functional_scaffold.core.metrics_unified import observe, reset_metrics_manager
+        from functional_scaffold.core.metrics_unified import observe

-        reset_metrics_manager()
-        observe("http_request_duration_seconds", {"method": "GET", "endpoint": "/"}, 0.1)
+        await observe("http_request_duration_seconds", {"method": "GET", "endpoint": "/"}, 0.1)

        mock_instance.pipeline.assert_called()

@@ -160,42 +241,49 @@ class TestConvenienceFunctions:
 class TestExport:
    """导出功能测试"""

-    @patch("redis.Redis")
-    def test_export_counter(self, mock_redis_class):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_export_counter(self, mock_redis_class):
        """测试导出计数器"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
-        mock_instance.hgetall.return_value = {"method=GET,endpoint=/,status=success": "10"}
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+        mock_instance.hgetall = AsyncMock(
+            return_value={"method=GET,endpoint=/,status=success": "10"}
+        )
+        mock_instance.hget = AsyncMock(return_value="0")
+        mock_instance.close = AsyncMock()
        mock_redis_class.return_value = mock_instance

-        from functional_scaffold.core.metrics_unified import export, reset_metrics_manager
+        from functional_scaffold.core.metrics_unified import export

-        reset_metrics_manager()
-        output = export()
+        output = await export()

        assert "http_requests_total" in output
        assert "HELP" in output
        assert "TYPE" in output

-    @patch("redis.Redis")
-    def test_export_histogram(self, mock_redis_class):
+    @pytest.mark.asyncio
+    @patch("redis.asyncio.Redis")
+    async def test_export_histogram(self, mock_redis_class):
        """测试导出直方图"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
-        mock_instance.hgetall.side_effect = lambda key: (
-            {"method=GET,endpoint=/": "5"}
-            if "count" in key
-            else {"method=GET,endpoint=/": "0.5"}
-            if "sum" in key
-            else {}
-        )
-        mock_instance.hget.return_value = "3"
+        mock_instance = AsyncMock()
+        mock_instance.ping = AsyncMock(return_value=True)
+
+        async def mock_hgetall(key):
+            if "count" in key:
+                return {"method=GET,endpoint=/": "5"}
+            elif "sum" in key:
+                return {"method=GET,endpoint=/": "0.5"}
+            return {}
+
+        mock_instance.hgetall = mock_hgetall
+        mock_instance.hget = AsyncMock(return_value="3")
+        mock_instance.close = AsyncMock()
        mock_redis_class.return_value = mock_instance

-        from functional_scaffold.core.metrics_unified import export, reset_metrics_manager
+        from functional_scaffold.core.metrics_unified import export

-        reset_metrics_manager()
-        output = export()
+        output = await export()

        assert "http_request_duration_seconds" in output

@@ -226,21 +314,9 @@ class TestEnvVarSubstitution:
 class TestTrackAlgorithmExecution:
    """track_algorithm_execution 装饰器测试"""

-    @patch("redis.Redis")
-    def test_decorator_success(self, mock_redis_class):
+    def test_decorator_success(self):
        """测试装饰器成功执行"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
-        mock_pipeline = MagicMock()
-        mock_instance.pipeline.return_value = mock_pipeline
-        mock_redis_class.return_value = mock_instance
-
-        from functional_scaffold.core.metrics_unified import (
-            reset_metrics_manager,
-            track_algorithm_execution,
-        )
-
-        reset_metrics_manager()
+        from functional_scaffold.core.metrics_unified import track_algorithm_execution

        @track_algorithm_execution("test_algo")
        def test_func():
@@ -249,21 +325,9 @@ class TestTrackAlgorithmExecution:
        result = test_func()
        assert result == "result"

-    @patch("redis.Redis")
-    def test_decorator_error(self, mock_redis_class):
+    def test_decorator_error(self):
        """测试装饰器错误处理"""
-        mock_instance = MagicMock()
-        mock_instance.ping.return_value = True
-        mock_pipeline = MagicMock()
-        mock_instance.pipeline.return_value = mock_pipeline
-        mock_redis_class.return_value = mock_instance
-
-        from functional_scaffold.core.metrics_unified import (
-            reset_metrics_manager,
-            track_algorithm_execution,
-        )
-
-        reset_metrics_manager()
+        from functional_scaffold.core.metrics_unified import track_algorithm_execution

        @track_algorithm_execution("test_algo")
        def test_func():
Author	SHA1	Message	Date
Roog (顾新培)	b47be9dda4	main:新增健康检查支持和服务优化 - 在 Worker 中引入轻量级 HTTP 服务器，支持健康检查和就绪检查端点。 - 在 Kubernetes 和 Docker 配置中新增健康检查探针，提升服务稳定性。 - 更新依赖，引入 `aiohttp` 用于健康检查服务。 - 优化部署配置，调整 Redis 主机配置和镜像地址以适配新环境。	2026-02-04 12:00:30 +08:00
Roog (顾新培)	55419443cd	main:新增健康检查支持和服务优化 - 在 Worker 中引入轻量级 HTTP 服务器，支持健康检查和就绪检查端点。 - 在 Kubernetes 和 Docker 配置中新增健康检查探针，提升服务稳定性。 - 更新依赖，引入 `aiohttp` 用于健康检查服务。 - 优化部署配置，调整 Redis 主机配置和镜像地址以适配新环境。	2026-02-04 11:58:56 +08:00
Roog (顾新培)	e0138d5531	main:新增阿里云 FC 部署文档及相关配置 - 更新 README，添加阿里云 FC 部署文档的链接。 - 新增 `docs/fc-deploy.md`，提供 FC 服务部署指南，包括环境准备与操作步骤。 - 优化文档表格格式，增加内容的可读性与完整性。	2026-02-04 11:36:01 +08:00
Roog (顾新培)	c92cac6ebb	main:完善 Redis 密码配置支持 - 在函数计算配置文件中新增 `redis_password` 字段。 - 更新 API 和 Worker 环境变量以传递 Redis 密码。 - 提升服务安全性，支持连接受保护的 Redis 实例。	2026-02-04 11:24:29 +08:00
Roog (顾新培)	c76ece8f48	main:移除无效 Docker 镜像配置 - 从 `docker-compose.yml` 中删除无效的镜像配置，以简化服务环境设置。	2026-02-04 10:39:40 +08:00
Roog (顾新培)	d211074576	main:更新阿里云函数计算配置为 FC 3.0 变更内容： - 重构函数计算配置文件，移除旧版 aliyun-fc.yaml，新增符合 FC 3.0 标准的 s.yaml。 - 引入 Serverless Devs 工具支持，添加部署、验证、日志查看等命令指引。 - 调整 API 和 Worker 函数配置，支持更灵活的资源分配及自动化管理。 - 更新文档，提供 FC 3.0 部署指南及优化建议。	2026-02-04 10:27:01 +08:00
Roog (顾新培)	a4d2ad1e93	main:采用异步 Redis 客户端优化指标管理模块变更内容： - 将 `redis` 客户端替换为 `redis.asyncio` 实现。 - 系统中同步方法调整为异步方法，提升事件循环效率。 - 在 `MetricsManager` 中添加异步初始化及关闭逻辑，避免阻塞问题。 - 更新便捷函数以支持异步上下文，并添加同步模式的兼容方法。 - 调整 Worker、JobManager、API 路由等模块，适配异步指标操作。 - 扩展单元测试，覆盖新增的异步方法及 Redis 操作逻辑。 - 简化 Dockerfile，取消开发依赖安装命令。	2026-02-03 19:54:22 +08:00
Roog	b5ca0e0593	Initial commit	2026-02-03 19:54:20 +08:00