One Time即一次性任务调度,配置完成后任务将在设置的时间点执行一次,执行完成后调度平台会自动清理任务,无需在客户端执行删除任务的操作。一次性任务调度适用于订单超时未支付、自动关闭定时日历提醒等场景。本文介绍如何通过创建一次性调度任务。
优势
精准时刻:SchedulerX一次性任务相较于延时消息,无固定延迟时间或执行时间限制,支持在未来的任意时间点执行,且使用简单。
丰富的任务类型:SchedulerX一次性任务适用于所有任务类型,包括Java、HTTP、Shell任务,并适用于所有分布式模型,例如单机、广播、分片、MapReduce等。
可视化运维:SchedulerX一次性任务与其他任务一致,具有可视化界面,方便观测和查询。并且支持在调度时间到达之前修改任务参数,支持失败自动重试等功能。
One Time定时调度配置示例
在创建调度任务时可选择One Time定时调度。
进入定时配置步骤面板,详情请参见创建调度任务。
在创建任务配置向导中完成基本配置,单击下一步进入定时配置面板。
在时间类型右侧的列表中选择one_time,并配置调度时间。
可选:如果需要,在高级设置中可设置时间偏移、时区和生效时间。
配置名称
意义
默认值
数据时间偏移
数据时间偏移,单位为秒。
无
时区
根据实际情况选择。
无
生效时间
指定任务开始生效时间。
立即生效
查看任务
登录MSE SchedulerX控制台,并在顶部菜单栏选择地域。
在左侧导航栏单击任务管理,可查看创建的定时调度任务。
通过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:])
该文章对您有帮助吗?