One Time

One Time即一次性任务调度,配置完成后任务将在设置的时间点执行一次,执行完成后调度平台会自动清理任务,无需在客户端执行删除任务的操作。一次性任务调度适用于订单超时未支付、自动关闭定时日历提醒等场景。本文介绍如何通过创建一次性调度任务。

优势

  • 精准时刻:SchedulerX一次性任务相较于延时消息,无固定延迟时间或执行时间限制,支持在未来的任意时间点执行,且使用简单。

  • 丰富的任务类型:SchedulerX一次性任务适用于所有任务类型,包括Java、HTTP、Shell任务,并适用于所有分布式模型,例如单机、广播、分片、MapReduce等。

  • 可视化运维:SchedulerX一次性任务与其他任务一致,具有可视化界面,方便观测和查询。并且支持在调度时间到达之前修改任务参数,支持失败自动重试等功能。

One Time定时调度配置示例

在创建调度任务时可选择One Time定时调度。

  1. 进入定时配置步骤面板,详情请参见创建调度任务

  2. 创建任务配置向导中完成基本配置,单击下一步进入定时配置面板。

  3. 时间类型右侧的列表中选择one_time,并配置调度时间

    可选:如果需要,在高级设置中可设置时间偏移时区生效时间

    配置名称

    意义

    默认值

    数据时间偏移

    数据时间偏移,单位为秒。

    时区

    根据实际情况选择。

    生效时间

    指定任务开始生效时间。

    立即生效

查看任务

  1. 登录MSE SchedulerX控制台,并在顶部菜单栏选择地域。

  2. 在左侧导航栏单击任务管理,可查看创建的定时调度任务。

一次性任务

通过API创建一次性调度任务

下面以一段示例代码说明如何使用SDK调用API,将timeType的参数值设置为5,表示启用一次性任务调度。详情请参见OpenAPI门户

import os
import sys

from typing import List

from alibabacloud_schedulerx220190430.client import Client as schedulerx220190430Client
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_schedulerx220190430 import models as schedulerx_220190430_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> schedulerx220190430Client:
        """
        使用凭据初始化账号Client
        @return: Client
        @throws Exception
        """
        # 工程代码建议使用更安全的无AK方式,凭据配置方式请参见:https://help.aliyun.com/document_detail/378659.html。                       
        config = open_api_models.Config(
            access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')
            access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
        )
        # Endpoint 请参考 https://api.aliyun.com/product/schedulerx2
        config.endpoint = f'schedulerx.aliyuncs.com'
        return schedulerx220190430Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        create_job_request = schedulerx_220190430_models.CreateJobRequest(
            region_id='cn-hangzhou',
            time_type=5,
            namespace='YOUR_NAMESPACE',
            group_id='YOUR_GROUP_ID',
            job_type='python',
            name='testjob',
            execute_mode='standalone',
            time_expression='2025-10-10 10:10:00',
            class_name='com.alibaba.schedulerx.test.helloworld'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            client.create_job_with_options(create_job_request, runtime)
        except Exception as error:
            # 此处仅做打印展示,请谨慎对待异常处理,在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        create_job_request = schedulerx_220190430_models.CreateJobRequest(
            region_id='cn-hangzhou',
            time_type=5,
            namespace='YOUR_NAMESPACE',
            group_id='YOUR_GROUP_ID',
            job_type='python',
            name='testjob',
            execute_mode='standalone',
            time_expression='2025-10-10 10:10:00',
            class_name='com.alibaba.schedulerx.test.helloworld'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            await client.create_job_with_options_async(create_job_request, runtime)
        except Exception as error:
            # 此处仅做打印展示,请谨慎对待异常处理,在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])