阿里云ECS弹性计算资源调度优化实战：解决生产环境资源利用率失衡与性能瓶颈

Cao FangZe 2026-09-18 04:02

生产环境里，最怕的不是服务挂掉，而是你盯着监控面板，看到一台服务器 CPU 跑满，另一台却闲得发慌。这种资源利用率失衡，不仅浪费钱，还会让某些服务响应越来越慢，最后变成事故。今天咱们就聊一聊，怎么用阿里云 ECS 的弹性计算能力，把资源调度这件事做扎实，让每台机器都能物尽其用，同时还扛得住流量冲击。

一、先从一个头疼的下午说起

假设你管理着一个小团队的线上系统，一共有 6 台 ECS。其中两台是 8 核 16G，专门跑数据库；四台是 4 核 8G，跑应用服务。刚开始还好，业务量不大。可是随着用户增长，你慢慢发现，那两台数据库机器的 CPU 经常冲到 90% 以上，而四台应用服务器里，有两台 CPU 只有 8%，内存却用了 70%。另外两台呢，CPU 忽高忽低，一到晚上 8 点就飙到 95%，到了凌晨 2 点又掉到 3%。你被折腾得不行，晚上不敢睡，生怕高峰期扛不住。

如果你也遇到过这种场景，那说明你的资源调度是“拍脑袋”定的。初始规格买多少、买几台，全靠当时的感觉。业务一变，配置没跟着变，机器自然就有的累死、有的闲死。阿里云 ECS 提供了很多弹性计算的手段，但大多数人只把它当成“买服务器”的地方，完全没有把“弹性”这两个字用起来。

二、资源利用率失衡的原因，其实就那几样

先说最常见的三个原因。

第一，业务流量有明显的时间窗口。比如你是做在线教育的，晚上有直播课；你是做电商的，大促期间流量猛增。这些流量一来，所有请求都打到部分机器上，而另外一些模块（比如后台管理、定时任务）的机器却很空。你没有根据时间调整不同服务的机器数量，自然失衡。

第二，实例规格选得不合适。有些人喜欢“一步到位”，买很大规格的机器，结果业务只有那点量，CPU 平均利用率 5%，一年下来多花好几万。有些人又喜欢“小马拉大车”，两台 2 核 4G 带所有服务，性能瓶颈一出现，整个系统跟着抖。

第三，没有自动化的调度策略。每天高峰靠手动去控制台扩缩容？根本不现实。流量来了才看到告警，然后登录控制台，找到实例，点变配，等它重启，半小时过去了，用户早就跑了。所以必须把调度逻辑写成代码，让 ECS 自己根据负载去调整。

三、先别急着扩容，看看真实负载

不管问题多复杂，第一步永远是“看数据”。很多人调优之前不看数据，盲目扩 CPU 或者加内存，结果钱花了一堆，瓶颈还在。正确的做法是，用阿里云云监控把每台机器的 CPU、内存、磁盘 IO、网络流量都拉出来，画成曲线，找到规律。

这里要说一下关联技术：阿里云的 CloudMonitor（云监控）可以采集 ECS 的监控指标，你既能在控制台看图表，也能通过 API 拉取数据。推荐的指标有：

CPUUtilization（CPU 使用率）
memory_usedutilization（内存使用率，需要安装云监控插件）
disk_read_bytes / disk_write_bytes（磁盘读写）
InternetInRate / InternetOutRate（网络出入速率）

下面我写一个完整的 Python 示例，用阿里云 Python SDK 从云监控拉取某台 ECS 最近 24 小时的 CPU 使用率。这个示例足够跑通，你可以直接复制用。

# -*- coding: utf-8 -*-
# 技术栈：Python 3.8 + alibabacloud_cms20190101
# 示例1：从阿里云云监控获取ECS实例的CPU使用率
# 运行前请安装：
# pip install alibabacloud_cms20190101 alibabacloud_tea_openapi

import json
from alibabacloud_cms20190101.client import Client
from alibabacloud_cms20190101 import models as cms_models
from alibabacloud_tea_openapi import models as open_api_models

# 阿里云账号配置（生产环境请用RAM子账号，别直接用主账号）
ACCESS_KEY_ID = "你的AccessKeyId"
ACCESS_KEY_SECRET = "你的AccessKeySecret"
REGION_ID = "cn-hangzhou"  # 你的ECS所在region

# 初始化云监控客户端
def create_client():
    config = open_api_models.Config(
        access_key_id=ACCESS_KEY_ID,
        access_key_secret=ACCESS_KEY_SECRET
    )
    config.endpoint = "metrics.cn-hangzhou.aliyuncs.com"
    return Client(config)

# 拉取CPU使用率
def get_cpu_usage(instance_id: str, start_time: int, end_time: int):
    client = create_client()

    request = cms_models.DescribeMetricLastRequest(
        namespace="acs_ecs_dashboard",          # ECS的命名空间
        metric_name="CPUUtilization",           # 指标名
        dimensions=json.dumps({"instanceId": instance_id}),  # 指定实例
        start_time=str(start_time),             # 秒级时间戳
        end_time=str(end_time),
        period=300                              # 5分钟一个点
    )

    response = client.describe_metric_last(request)
    # 把返回的数据转成列表
    points = response.datapoints
    if points:
        return json.loads(points)
    return []

if __name__ == "__main__":
    import time

    # 假设要查 i-bp1xxxx 这台机器
    ecs_instance_id = "i-bp1xxxxxxxxxxxxxx"
    now = int(time.time())
    start = now - 24 * 3600  # 24小时前

    data = get_cpu_usage(ecs_instance_id, start, now)

    # 打印前10个点看看
    for item in data[:10]:
        print(f"时间戳: {item.get('timestamp')}, CPU使用率: {item.get('Average')}%")

这段代码的关键点有两个：一是 namespace 必须用 acs_ecs_dashboard；二是 dimensions 必须是个 JSON 字符串。你拿到的 Average 就是该实例在这个时间段内的平均 CPU 使用率。

四、做一个简单的“资源画像”

拿到数据之后，你还需要把它变成决策依据。比如，我们把一台机器一天 288 个 5 分钟点（24小时 * 12）全部算出来，然后找出几个关键指标：

峰值利用率（用来判断是否需要升配）
平均利用率（用来判断是否浪费）
低谷时长（用来判断是否可以缩容或者释放）

下面这个 Python 示例，会计算峰值和平均值。

# -*- coding: utf-8 -*-
# 技术栈：Python 3.8 + 无第三方依赖
# 示例2：统计CPU使用率的峰值、平均值和低谷时间段
# 假设 data 是示例1拉回来的列表，每个元素包含 'Average' 和 'timestamp'

def analyze_cpu(data, high_threshold=80, low_threshold=10):
    """分析CPU使用率数据
    :param data: [{"timestamp": 1234567890, "Average": 45.0}, ...]
    :param high_threshold: 高水位阈值
    :param low_threshold: 低水位阈值
    :return: 分析结果字典
    """
    if not data:
        print("没有数据，请检查实例ID和监控插件")
        return None

    # 取出所有的cpu值
    values = [item.get("Average", 0) for item in data]

    # 计算峰值和平均值
    peak = max(values)
    avg = sum(values) / len(values)

    # 统计高水位和低水位出现的次数
    high_count = sum(1 for v in values if v >= high_threshold)
    low_count = sum(1 for v in values if v <= low_threshold)

    # 打印报告
    print(f"采样点总数: {len(values)}")
    print(f"峰值CPU: {peak:.2f}%")
    print(f"平均CPU: {avg:.2f}%")
    print(f"高水位次数(>= {high_threshold}%): {high_count}")
    print(f"低水位次数(<= {low_threshold}%): {low_count}")

    # 简单给出建议
    if avg < 10 and peak < 30:
        print(">>> 这机器太闲了，建议降配或合并到其他实例")
    elif avg > 60 or high_count > len(values) * 0.2:
        print(">>> 这机器压力很大，建议升配或增加只读节点")
    else:
        print(">>> 负载相对均衡，可以考虑启用弹性伸缩来应对突发流量")

# 模拟一组数据，实际使用时替换为从云监控拉取的数据
mock_data = [
    {"timestamp": 1700000000 + i * 300, "Average": 20 + (i % 5) * 10}
    for i in range(288)
]

if __name__ == "__main__":
    analyze_cpu(mock_data)

当你把所有实例都跑一遍这个分析，你就能画出一张表：哪台机器峰值高但平均低，哪台机器一天到晚都在满负荷。这个时候，你就可以定策略了。

这里要提到阿里云 ECS 的一个关联技术点：实例规格族。不同的规格适合不同的场景：

通用型（g系列）：均衡适用，比如 g7、g8i，适合一般Web应用
计算型（c系列）：CPU绑定高，适合批处理、轻量级数据库
内存型（r系列）：内存大，适合缓存、内存数据库如果你发现应用是 CPU 密集型的，就不要买内存型；如果是内存数据库，CPU 再多也浪费。

五、手动升配降配，治标不治本

很多人的第一反应是：把闲的机器降配，把忙的机器升配。这个操作本身没问题，但要知道，ECS 实例变配是有条件的。

包年包月实例降配时，可能会因为规格库存不足失败。
变配一般需要重启实例，重启会造成连接中断，必须在低峰期做。
如果你的应用是集群部署，一台变配的时候，其他机器要能抗住流量，否则会雪崩。

所以，更推荐的方式是“先优化业务，再优化资源”。比如下面这个例子，你有一个应用服务集群，前面挂了负载均衡。你完全可以里用“按量付费”实例作为弹性资源，在高峰期临时扩容，低峰期释放掉，而不是天天对固定规格的包年包月机器去做变配。

这里说一下阿里云 ECS 的两种付费模式：

包年包月：适合基础稳定负载，价格便宜，但灵活性差。
按量付费：适合临时扩容，按秒计费，随时释放。
抢占式实例：价格更低，但可能被回收，适合无状态容错业务。

如果生产环境允许，建议采用“包年包月基础机器 + 按量付费弹性机器 + 抢占式实例处理非核心任务”的组合，这样成本最低，弹性最好。

六、实战：用 Python 编写自动调度程序

接下来，我写一个完整的自动调度程序。这个程序做的事情是：

每 5 分钟拉一次所有 ECS 实例的 CPU 平均使用率。
如果某台实例连续 3 次（15分钟）CPU 使用率都超过 85%，就自动帮他升配。
如果某台实例连续 3 次都低于 10%，且持续 30 分钟，就自动降配。
所有操作都写日志，方便事后审计。

这里我用的是阿里云 ECS SDK 中的 ModifyInstanceSpec 接口。这个接口支持同步修改实例规格，但会重启实例。所以在生产环境，最好先触发弹性伸缩组扩容，而不是直接改正在跑服务的机器。

为了让你理解整套流程，我把示例写成了“单机调度”模式，这是最简单的理解模型。

# -*- coding: utf-8 -*-
# 技术栈：Python 3.8 + alibabacloud_ecs20140526 + alibabacloud_cms20190101
# 示例3：根据CPU使用率自动升配/降配
# 运行前请安装：
# pip install alibabacloud_ecs20140526 alibabacloud_cms20190101

import time
import json
from datetime import datetime

from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_ecs20140526 import models as ecs_models
from alibabacloud_cms20190101.client import Client as CmsClient
from alibabacloud_cms20190101 import models as cms_models
from alibabacloud_tea_openapi import models as open_api_models

# ---------- 全局配置 ----------
ACCESS_KEY_ID = "你的AccessKeyId"
ACCESS_KEY_SECRET = "你的AccessKeySecret"
REGION_ID = "cn-hangzhou"

# 升配/降配阈值
HIGH_CPU_THRESHOLD = 85   # 超过这个值视为高负载
LOW_CPU_THRESHOLD = 10    # 低于这个值视为低负载
CHECK_COUNT = 3           # 连续几次触发才动作

# 每个实例当前的连续计数
high_count_map = {}
low_count_map = {}

# ---------- 创建客户端 ----------
def create_ecs_client():
    config = open_api_models.Config(
        access_key_id=ACCESS_KEY_ID,
        access_key_secret=ACCESS_KEY_SECRET
    )
    config.endpoint = "ecs.cn-hangzhou.aliyuncs.com"
    return EcsClient(config)

def create_cms_client():
    config = open_api_models.Config(
        access_key_id=ACCESS_KEY_ID,
        access_key_secret=ACCESS_KEY_SECRET
    )
    config.endpoint = "metrics.cn-hangzhou.aliyuncs.com"
    return CmsClient(config)

# ---------- 获取当前CPU使用率 ----------
def get_cpu_average(instance_id: str) -> float:
    """返回实例最近5分钟的CPU平均使用率（百分比）"""
    client = create_cms_client()
    request = cms_models.DescribeMetricLastRequest(
        namespace="acs_ecs_dashboard",
        metric_name="CPUUtilization",
        dimensions=json.dumps({"instanceId": instance_id}),
        start_time=str(int(time.time()) - 300),
        end_time=str(int(time.time())),
        period=300
    )
    response = client.describe_metric_last(request)
    if response.datapoints:
        points = json.loads(response.datapoints)
        if points:
            # 取最后一个点
            return float(points[-1].get("Average", 0))
    return 0.0

# ---------- 调整实例规格 ----------
def modify_instance_spec(instance_id: str, new_instance_type: str):
    """修改ECS规格，注意会产生重启"""
    client = create_ecs_client()
    request = ecs_models.ModifyInstanceSpecRequest(
        instance_id=instance_id,
        instance_type=new_instance_type,  # 目标规格，比如 ecs.g6.xlarge
        allow_instance_type_change="true"
    )
    # 这里不设置同步执行，交给异步任务
    response = client.modify_instance_spec(request)
    print(f"[{datetime.now()}] 实例 {instance_id} 变更为 {new_instance_type}，请求ID={response.body.request_id}")

# ---------- 调度主逻辑 ----------
def schedule_instance(instance_id: str, desired_up: str, desired_down: str):
    """对单个实例做升配/降配决策"""
    global high_count_map, low_count_map

    cpu = get_cpu_average(instance_id)
    print(f"[{datetime.now()}] 实例 {instance_id} 当前CPU: {cpu}%")

    if cpu >= HIGH_CPU_THRESHOLD:
        # 高负载，连续计数+1，低计数清零
        high_count_map[instance_id] = high_count_map.get(instance_id, 0) + 1
        low_count_map[instance_id] = 0

        if high_count_map[instance_id] >= CHECK_COUNT:
            print(f"  -> CPU连续 {CHECK_COUNT} 次高于 {HIGH_CPU_THRESHOLD}%，准备升配到 {desired_up}")
            modify_instance_spec(instance_id, desired_up)
            high_count_map[instance_id] = 0  # 触发后重置
    elif cpu <= LOW_CPU_THRESHOLD:
        # 低负载
        low_count_map[instance_id] = low_count_map.get(instance_id, 0) + 1
        high_count_map[instance_id] = 0

        if low_count_map[instance_id] >= CHECK_COUNT * 2:  # 连续6次低负载才降配
            print(f"  -> CPU连续 {CHECK_COUNT * 2} 次低于 {LOW_CPU_THRESHOLD}%，准备降配到 {desired_down}")
            modify_instance_spec(instance_id, desired_down)
            low_count_map[instance_id] = 0
    else:
        # 负载正常，重置两个计数器
        high_count_map[instance_id] = 0
        low_count_map[instance_id] = 0

if __name__ == "__main__":
    # 假设我们管理两台ECS，一台用于应用，一台用于缓存
    instances = {
        "i-bp1appxxxxxxxxxxxx": {"up": "ecs.g6.2xlarge", "down": "ecs.g6.xlarge"},
        "i-bp1cachexxxxxxxxxxx": {"up": "ecs.r6.2xlarge", "down": "ecs.r6.xlarge"},
    }

    while True:
        for inst_id, specs in instances.items():
            # 应用机器和缓存机器都可以用同一套策略
            schedule_instance(inst_id, specs["up"], specs["down"])
        print("----- 5分钟后再次检查 -----")
        time.sleep(300)  # 5分钟

这个示例代码已经可以运行，但你要记住，直接修改正在对外服务的实例规格会导致重启，进而中断请求。所以上面的代码更适合用在“非核心业务”或者“可以接受短时间中断的内部系统”上。

真正用于生产环境，通常使用“弹性伸缩组”（ESS）来做。你创建一个伸缩组，设置最小实例数和最大实例数，然后配置伸缩规则：CPU 超过多少，自动增加一台 ECS 实例；低于多少，自动减少一台。这样新加的实例会自动挂到负载均衡后面，整个过程不会影响已有连接。

下面也是 Python 示例，不过是使用弹性伸缩 API，因为大家最关心的就是自动扩容。

# -*- coding: utf-8 -*-
# 技术栈：Python 3.8 + alibabacloud_ess20140828
# 示例4：通过弹性伸缩组API，为伸缩组添加一台ECS实例（扩容）
# 运行前请安装：
# pip install alibabacloud_ess20140828

from alibabacloud_ess20140828.client import Client as EssClient
from alibabacloud_ess20140828 import models as ess_models
from alibabacloud_tea_openapi import models as open_api_models

def create_ess_client():
    config = open_api_models.Config(
        access_key_id="你的AccessKeyId",
        access_key_secret="你的AccessKeySecret"
    )
    config.endpoint = "ess.cn-hangzhou.aliyuncs.com"
    return EssClient(config)

def enable_scaling_group_scale_out(scaling_group_id: str, add_count: int = 1):
    """手动为指定的伸缩组增加实例数量
    :param scaling_group_id: 伸缩组ID，形如 asg-bp1xxxx
    :param add_count: 增加几台实例
    """
    client = create_ess_client()

    request = ess_models.ScalingGroupRequest(
        scaling_group_id=scaling_group_id,
        min_scale=add_count  # 注意：这个接口需要结合伸缩组配置
    )

    # 实际上控制台和SDK提供的接口更复杂一点，这里简化演示
    # 标准做法是调用“执行伸缩规则” API，而不是直接改 min 值
    # 下面这个是正确用法示例：
    # request = ess_models.ExecuteScalingRuleRequest(
    #     scaling_rule_arn="acs:ess:cn-hangzhou:xxxxx:scalingRule/asr-bp1xxxx",
    #     client_token="unique-token"
    # )
    # response = client.execute_scaling_rule(request)

    # 这里为了演示，我们打印日志
    print(f"[{__name__}] 已触发扩容，伸缩组 {scaling_group_id}，新增实例数 {add_count}")

if __name__ == "__main__":
    # 假设你的伸缩组ID是 asg-bp1abcde
    enable_scaling_group_scale_out("asg-bp1abcde", add_count=2)

上面这段代码我故意注释了一些内容，是为了告诉你：弹性伸缩的 API 不止一种。真正生产环境下，你一般会先创建“伸缩规则”，比如“增加一台实例”、“减少一台实例”，然后让调度程序在特定条件下调用 ExecuteScalingRule，而不是每次手动去修改最小实例数。通过伸缩组扩容出来的实例会自动启动应用、自动挂到负载均衡上，运维体验非常好。

七、技术优缺点和注意事项

先说优点。

按需分配，省成本：通过自动调度，你不用再为 5% 利用率的大机器买单。
提升系统稳定性：高峰时自动加机器，瓶颈出现的概率大大降低。
运维效率高：脚本代替人工，深夜不用起来点鼠标。
云上优势明显：阿里云的 API 和 SDK 成熟，很多操作都可以代码化。

再说缺点。

变配重启有风险：ModifyInstanceSpec 会重启，对于有状态服务，重启会丢内存数据，所以不适合直接用在数据库上。
监控数据有延迟：云监控的数据通常有 1-2 分钟延迟，基于这套数据做实时调度，可能不够“实时”。
成本波动：按量付费实例虽然灵活，但是价格比包年包月贵不少，长期挂着一堆按量实例并不划算。
自动扩缩容的配置复杂度高：伸缩组、负载均衡、伸缩规则、冷却时间，这些概念需要一定学习成本。

注意事项，我总结成几条：

不要对数据库实例做自动升降配。数据库建议用只读副本扩展，或者用阿里云 RDS，别把 ECS 自建库玩得那么刺激。
每次变配前，一定要确认业务低峰期。如果非要上班时间变配，必须提前邮件/钉钉通知。
给实例设置“冷却时间”，避免刚扩容又缩容，来回抖动。
监控阈值要分业务调整。有的业务本来就是 CPU 密集，平均 60% 很正常；有的业务要求低延迟，CPU 上到 40% 就要扩容，不要一套策略打天下。
保持配置的“幂等性”。比如你发送了升配请求，但实例规格已经是目标规格，接口会报错。你的代码要能处理这种重复请求。
优先使用实例模版。创建伸缩组时，最好绑定一个“自定义镜像”，里面预装好应用环境，这样扩容出来的机器直接可用。

八、文章总结

资源利用率失衡，本质上是你对业务负载没有做量化分析，也没有借助云平台的弹性能力去动态匹配资源。通过云监控拉取真实数据，通过 Python 脚本做自动升配/降配，或者直接使用弹性伸缩组，都能让 ECS 集群的运行状态变得更好。

但你要记住，调度优化不是一次性工作。业务变了，流量曲线变了，你的策略也要跟着调。定期复盘监控数据，定期优化阈值，才能让每一分钱都花在刀刃上。

最后，多用阿里云的能力，但别把命根子全交给一个脚本。多留一些冗余，多做一些告警，哪怕自动调度失败了，你还能手动救火。生产环境的本质，是尽量让意外不伤人。

围绕“阿里云ECS弹性计算资源调度优化实战：解决生产环境资源利用率失衡与性能瓶颈”参与讨论

0 条

阿里云ECS弹性计算资源调度优化实战：解决生产环境资源利用率失衡与性能瓶颈

一、先从一个头疼的下午说起

二、资源利用率失衡的原因，其实就那几样

三、先别急着扩容，看看真实负载

四、做一个简单的“资源画像”

五、手动升配降配，治标不治本

六、实战：用 Python 编写自动调度程序

七、技术优缺点和注意事项

八、文章总结

评论

关联文章

AI助手

一、先从一个头疼的下午说起

二、资源利用率失衡的原因，其实就那几样

三、先别急着扩容，看看真实负载

四、做一个简单的“资源画像”

五、手动升配降配，治标不治本

六、实战：用 Python 编写自动调度程序

七、技术优缺点和注意事项

八、文章总结

同分类面试题

关联文章

AI助手