Microservices Engine：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://www.alibabacloud.com/help/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:])