异步任务管理接口和设置

DashScope平台提供了一系列的API能力调用，其中部分接口因为相关能力需要比较长的处理时间或者一些其他的限制，采用了异步任务接口，即用户调用对应的能力API将会提交一个相关的异步任务，接口将立刻返回并告知客户任务提交的结果，如果提交成功，返回内容中将包含任务的序列号（task_id）；随后客户可以通过任务状态/结果查询任务的情况，相关的接口详细描述在对应的API详情文档中有详细说明。DashScope针对这一类的异步任务，提供了一系列通用的异步任务管理接口方便客户管控自己提交的异步任务，客户可以使用对应的API在需要的时候直接查询和管理队列中的异步任务。同时，除了提供DashScope的平台接口，异步任务模块还接入了阿里云事件总线（EventBridge），用户可以更加灵活地以消息等其他方式接收任务状态的更新信息。

异步任务管理接口

1. 异步任务查询接口（结果获取接口）

异步任务查询接口是一个全局的调用接口，用户在调用时候指定 task_id 即可获取到任务的运行状态，同时在任务结束（成功或者失败）的时候获取到任务的结果或者相应出错信息。

需要注意的是：

1. 用户能查询Apikey归属的阿里云主账号的所有任务（包括阿里云主账号下所有Apikey提交的任务），但无法查询到其他主账号下的任务；

2. 不同类型的异步任务对于已经结束的任务有不同的保留时间（通常来说这个时间为 24 小时，具体的请参考特定任务的相关文档），如果超出这个时间，则任务会被从系统中清除；

流量限制：

对于普通客户，异步任务查询接口提供 20 QPS 的访问流量限制，即每秒钟每个账号可以发起不超过20次的任务查询API调用，如果有超出基础限流的调用需求，可发送电子邮件至 dashscope@alibabacloud.com 额外申请。

入参描述：

出参描述：

请求示例：

curl -X GET 'https://dashscope.aliyuncs.com/api/v1/tasks/73205176-xxxx-xxxx-xxxx-16bd5d902219' \
--header 'Authorization: Bearer <YOUR_API_KEY>'

响应示例：

{
    "request_id": "45ac7f13-xxxx-xxxx-xxxx-e03c35068d83",
    "output": {
        "task_id": "73205176-xxxx-xxxx-xxxx-16bd5d902219",
        "task_status": "SUCCEEDED",
        "submit_time": "2023-12-20 21:36:31.896",
        "scheduled_time": "2023-12-20 21:36:39.009",
        "end_time": "2023-12-20 21:36:45.913",
        "results": [
            {
                "url": "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/xxx1.png"
            },
            {
                "url": "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/xxx2.png"
            },
            {
                "url": "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/xxx3.png"
            },
            {
                "code": "DataInspectionFailed",
                "message": "Output data may contain inappropriate content.",
            }
        ],
        "task_metrics": {
            "TOTAL": 4,
            "SUCCEEDED": 3,
            "FAILED": 1
        }
    },
    "usage": {
        "image_count": 3
    }
}

2. 异步任务取消接口

异步任务取消接口用于用户取消不再需要的任务。

需要注意的是：

1. 异步任务取消接口只能操作非运行状态和非结束状态的接口，如果任务一旦开始运行或者已经完成，则无法再行取消；

2. 取消接口能操作Apikey归属的阿里云主账号的所有任务（包括阿里云主账号下所有Apikey提交的任务），但无法操作到其他主账号下的任务。

流量限制：

对于普通客户，异步任务取消接口提供 20 QPS 的访问流量限制，即每秒钟每个账号可以发起不超过20次的任务取消API调用，如果有超出基础限流的调用需求，可发送电子邮件至 dashscope@alibabacloud.com 额外申请。

入参描述：

出参描述：

请求示例：

curl -X POST 'https://dashscope.aliyuncs.com/api/v1/tasks/73205176-xxxx-xxxx-xxxx-16bd5d902219/cancel' \
--header 'Authorization: Bearer <YOUR_API_KEY>'

响应示例：

{
    "request_id": "45ac7f13-xxxx-xxxx-xxxx-e03c35068d83"
}

3. 异步任务列表查询接口

查询用户提交的所有异步任务，该接口提供多种查询条件，用户可以通过相关条件的组合获取到自己的任务列表。异步任务列表查询接口和结果查询接口的限制条件一样，用户只能查询属于自己主账号的任务并且已经结束的任务在达到超时限制之后会被系统清除则无法再被列出。

流量限制：

对于普通客户，异步任务列表查询接口提供 20 QPS 的访问流量限制，即每秒钟每个账号可以发起不超过20次的任务列表查询API调用，如果有超出基础限流的调用需求，可发送电子邮件至 dashscope@alibabacloud.com 额外申请。

入参描述：

出参描述：

请求示例：

curl -X GET 'https://dashscope.aliyuncs.com/api/v1/tasks/?start_time=xxx&end_time=xxx&api_key_id=xxx&region=xxx&task_id=xxx&status=xxx' \
--header 'Authorization: Bearer <YOUR_API_KEY>'

响应示例：

{
    "request_id": "eeaf6e5e-xxxx-xxxx-xxxx-7b7c2d399c3f",
    "data": [
        {
            "api_key_id": "123",
            "gmt_create": 1682514589152,
            "gmt_create": 1682514589152,
            "caller_uid": "xxxxxxxxx",
            "start_time": 1682515862179,
            "end_time": 1682527200093,
            "model_name": "paraformer-16k-1",
            "region": "cn-hangzhou",
            "request_id": "32b67b58-xxxx-xxxx-xxxx-230f0aee64d9",
            "status": "FAILED",
            "task_id": "cf52b16b-xxxx-xxxx-xxxx-17f9c211440c",
            "user_api_unique_key": "apikey:v1:audio:asr:transcription:paraformer-16k-1"
        }
    ],
    "total_page": 4,
    "total": 39,
    "page_no": 1,
    "page_size": 10
}

异步任务外部拓展

为了满足客户的多元化需求，DashScope异步任务系统还接入了阿里云事件总线——EventBridge。

当异步任务完成后，DashScope异步任务系统会将相应的消息发送给EventBridge，用户可以在EventBridge控制台配置自己的事件消息处理机制，比如指定的http回调url、投递到指定的MQ队列，以便更加高效处理已完成的DashScope异步任务。

1. EventBridge配置事件转发

EventBridge的几个核心概念：

• 事件：事件状态变化的数据记录。

• 事件源：事件的来源，负责生产事件。

• 事件目标：事件的处理终端，负责消费事件。

• 事件总线：事件的中转站，负责事件的中间转储。

• 事件规则：用于监控特定类型的事件。当发生匹配事件时，事件会被路由到与事件规则关联的事件目标。

详细信息，请参见EventBridge帮助文档。

云控制台，请登录EventBridge控制台。

EventBridge中配置事件接收的准备工作：

查询DashScope异步任务事件

DashScope异步任务结束后，在EventBridge中可以查询DashScope投递的任务结束事件。

登录阿里云主账号，然后进入EventBridge控制台，切换到北京地域，选择左侧事件总线导航，点击default进入云服务专用总线。

DashScope所属的事件总线为云服务专用总线——即defalut。

点击事件追踪，输入查询条件，查询DashScope任务结束事件。

• 事件源：acs.dashscope

• 事件类型：dashscope:System:AsyncTaskFinish

点击事件详情，可以查看到DashScope任务结束的详细信息：

{
    "datacontenttype": "application/json;charset=utf-8",
    "aliyunaccountid": "xxxxx",
    "aliyunpublishtime": "2023-10-25T01:45:16.993Z",
    "data": {
        "start_time": "2023-10-25 09:45:09",
        "user_api_unique_key": "apikey:v1:audio:asr:transcription:paraformer-8k-v1",
        "task_status": "SUCCEEDED",
        "end_time": "2023-10-25 09:45:16",
        "task_id": "a154c328-xxxx-xxxx-xxxx-e52a9a7e9a35",
        "region": "cn-beijing",
        "request_id": "108f38f5-xxxx-xxxx-xxxx-6504db9080b3",
        "api_key_id": "1250"
    },
    "aliyunoriginalaccountid": "xxxxxxxx",
    "specversion": "1.0",
    "aliyuneventbusname": "default",
    "id": "81765e5b-xxxx-xxxx-xxxx-bbad8dde2bd9",
    "source": "acs.dashscope",
    "time": "2023-1-25T01:45:16.969Z",
    "aliyunregionid": "cn-beijing",
    "type": "dashscope:System:AsyncTaskFinish"
}

参数描述：

创建转发规则

这里可以自定义规则，将DashScope任务转发到不同的接收点。

规则名称和描述可以自定义

配置事件模式

需要选择DashScope的事件源和DashScope任务结束的事件类型。

• 事件源：acs.dashscope 代表DashScope云产品

• 事件类型：dashscope:System:AsyncTaskFinish 代表DashScope 异步任务结束

• 模式内容：用来过滤相关事件的模式定义，可以根据指定字段的值进行过滤，具体模式定义参考事件模式

例如，只转发模型名称为paraformer-8k-v1的事件：

{
  "source": ["acs.dashscope"],
  "type": ["dashscope:System:AsyncTaskFinish"],
  "data": {
    "user_api_unique_key": [
      {"suffix": ":paraformer-8k-v1"}
    ]
  }
}

配置事件目标

EventBridge支持多种类型的事件目标，包括http、RocketMQ消息队列（事件目标概述）

以下给出两个最常用的事件接收方式，其他接收方式可以参考事件目标参数

2. HTTP回调接口

使用http接口的形式接收EventBridge事件，需要提供一个公网或者阿里云专有网络的http的服务

打开配置EventBridge的事件目标界面

服务类型：选择http

URL：填写http服务地址

Body：完整事件

网络类型：根据服务地址选择

http支持公网和专用网络两种类型，当选择专用网络时，需要配置VPC、vSwitch和SecurityGroup信息。

如果选择完整事件，事件总线 EventBridge 将会把全部的事件内容路由到目标。点击事件内容转换查看使用示例。

点击确认即可完成规则的修改，查看事件目标，如果有http样式，则代表配置成功

3. RocketMQ消息队列

通过RocketMQ消息队列接收EventBridge事件，需要先准备好RocketMQ队列。

进入RocketMQ的管理控制台，创建RocketMQ的实例（如果已经有RocketMQ队列，可以跳过此步）

下图中，rmq-cn-nwy*******为实例ID

创建对应实例的Topic和Group

下图中，rmq-cn-nwy*******为实例ID

RocketMQ创建完成后，打开配置EventBridge的事件目标界面，选择已配置的RocketMQ实例。

服务类型：消息队列RocketMQ版

版本：创建完成的RocketMQ版本

实例ID：创建完成的RocketMQ的实例ID

Topic：创建完成的Topic

配置完成后，提交异步任务，待任务完成后，在配置的RocketMQ的Topic中查看消息

RocketMQ在线查看消息需要 开通消息一键收发体验功能

消息一键收发体验功能是基于函数计算的产品能力实现的，运行 SDK 示例如果超过了计费概述会产生少量费用，点击计费概述查看函数计算收费规则。

通过RocketMQ的SDK，接收消息。

SDK参考

RocketMQ 5.0版本，java客户端代码实例如下：

Maven项目中，引入以下依赖：

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client-java</artifactId>
  <version>5.0.4</version>
</dependency>

消费MQ消息代码：

package com.test;

import com.alibaba.fastjson2.JSON;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;

/**
 * @author xxx
 */
public class ConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerExample.class);

    private ConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        /*
          实例接入点，从控制台实例详情页的接入点页签中获取。
          如果是在阿里云ECS内网访问，建议填写VPC接入点。
          如果是在本地公网访问，或者是线下IDC环境访问，可以使用公网接入点。使用公网接入点访问，必须开启实例的公网访问功能。
         */
        String endpoints = "xxxx";
        //指定需要订阅哪个目标Topic，Topic需要提前在控制台创建，如果不创建直接使用会返回报错。
        String topic = "xxxx";
        //为消费者指定所属的消费者分组，Group需要提前在控制台创建，如果不创建直接使用会返回报错。
        String consumerGroup = "xxxx";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /*
          如果是使用公网接入点访问，configuration还需要设置实例的用户名和密码。用户名和密码在控制台实例详情页获取。
          如果是在阿里云ECS内网访问，无需填写该配置，服务端会根据内网VPC信息智能获取。
         */
        builder.setCredentialProvider(new StaticSessionCredentialsProvider("xxxx", "xxxx"));
        ClientConfiguration clientConfiguration = builder.build();
        //订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        //初始化SimpleConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        .setClientConfiguration(clientConfiguration)
        //设置消费者分组。
        .setConsumerGroup(consumerGroup)
        //设置预绑定的订阅关系。
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        //设置消费监听器。
        .setMessageListener(messageView -> {
            try {
                //处理消息并返回消费结果。
                ByteBuffer buffer = messageView.getBody();
                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity());
                for (int i = 0; i < buffer.capacity(); i++) {
                    newBuffer.put(buffer.get(i));
                }
                String result = new String(newBuffer.array());
                LOGGER.info("Consume message={}", JSON.toJSONString(result));
                System.out.println(result);
                return ConsumeResult.SUCCESS;
            } catch (Exception e) {
                LOGGER.error("deal message has error", e);
                return ConsumeResult.FAILURE;
            }
        })
        .build();
        Thread.sleep(Long.MAX_VALUE);

        //如果不需要再使用PushConsumer，可关闭该进程。
        pushConsumer.close();
    }
}                                                 

4. 常见问题

1. 配置完成了事件规则，但是接收不到事件？

检查事件规则配置的地域，以及事件目标所属地域是否一致

2. 一个事件规则可以配置多个事件目标吗？

可以，同一个事件规则可以配置多个事件目标。如果配置多个事件目标，则同一个事件会投递到配置的每个事件目标中。

3. HTTP/HTTPS服务请求超时或者请求错误？

检查HTTP/HTTPS服务状态；

检查事件目标配置的url是否正确；

检查事件目标配置的Network类型：

PublicNetwork：公网；

PrivateNetwork：VPC网络，如果选择此项，需要配置VPC、vSwitch和SecurityGroup信息；

检查VPC网络和交换机配置信息是否正确。检查网络安全组配置是否正确。

更多参数配置请参考：事件目标参数