DataWorks的OpenEvent為您提供消息推送訂閱的能力，同時，您可將服務程序注冊為DataWorks的擴展程序，通過擴展程序來卡點并響應訂閱的事件消息，實現通過擴展程序對特定事件進行消息通知與流程管控。詳情請參見擴展程序概述。

1. 開啟并配置消息訂閱（OpenEvent）。

   • 由于“查詢結果下載”為非工作空間內的操作，因此本案例將使用Default總線來承接操作事件消息。

     說明

     如需通過RAM子賬號/RAM角色身份讀取Default總線中的事件，請提前進行授權。

     

   • 查詢結果下載的事件名稱為“dataworks:ResourcesDownload:DownloadResources”，可參考下圖查看詳情。

     

     事件詳情及關鍵參數含義。

     {
       "datacontenttype": "application/json;charset=utf-8",
       "aliyunaccountid": "1107550004253538",
       "aliyunpublishtime": "2023-12-05T07:25:31.708Z",
       "data": {
         "eventCode": "download-resources",
         "extensionBizId": "audit_4d7ebb42b805428483148295a97a8404",
         "extensionBizName": "DataWorks_IDE_Query_20231205152530.csv",
         "requestId": "77cac0c2fc12cecbf1d289128897cc7b@@ac15054317017611303051804e8b43",
         "appId": 3159,
         "tenantId": 524257424564736,
         "blockBusiness": true,
         "eventBody": {
           "sqlText": "SELECT * FROM table_1",   // 查詢SQL
           "queryDwProjectId": "3159",    // 查詢數據源所在的工作空間ID
           "moduleType": "develop_query",  // 下載來源：develop_query（數據開發查詢）/sqlx_query（數據分析查詢）/dw_excel（數據分析電子表格）
           "operatorBaseId": "1107550004253538", //操作者的UID
           "datasourceId": "18889", //查詢的數據源ID
           "queryDwProjectName": "yongxunQA_emr_chengdu1", // 查詢數據源所在的工作空間名稱
           "dataRowSize": 4577,  // 下載的數據量
           "datasourceName": "odps_source",  // 查詢的數據源名稱
           "operatorUid": "1107550004253538"
         },
         "operator": "1107550004253538"
       },
       "aliyunoriginalaccountid": "1107550004253538",
       "specversion": "1.0",
       "aliyuneventbusname": "default",
       "id": "169d171c-d523-4370-a874-bb0fa083194d",
       "source": "acs.dataworks",
       "time": "2023-12-05T15:25:31.588Z",
       "aliyunregionid": "cn-chengdu",
       "type": "dataworks:ResourcesDownload:DownloadResources"
     }

2. 開發部署線擴展程序（Extensions）。

   說明

   實際使用時，請結合上一步消息事件內容進行風險判斷。

   處理的擴展點選擇“數據下載前置事件”。

   開發擴展程序可參考的示例代碼。

   package com.aliyun.dataworks.demo;
   
   import com.alibaba.fastjson.JSON;
   import com.alibaba.fastjson.JSONObject;
   import com.aliyun.dataworks.config.Constants;
   import com.aliyun.dataworks.config.EventCheckEnum;
   import com.aliyun.dataworks.config.ExtensionParamProperties;
   import com.aliyun.dataworks.services.DataWorksOpenApiClient;
   import com.aliyuncs.IAcsClient;
   import com.aliyuncs.dataworks_public.model.v20200518.*;
   import com.aliyuncs.exceptions.ClientException;
   import org.springframework.beans.factory.annotation.Autowired;
   import org.springframework.web.bind.annotation.*;
   
   /**
    * @author dataworks demo
    */
   @RestController
   @RequestMapping("/extensions")
   public class ExtensionsController {
   
       @Autowired(required = false)
       private DataWorksOpenApiClient dataWorksOpenApiClient;
   
       @Autowired
       private ExtensionParamProperties extensionParamProperties;
   
       /**
        * 接收eventBridge推送過來的消息
        * @param jsonParam
        */
       @PostMapping("/consumer")
       public void consumerEventBridge(@RequestBody String jsonParam){
           JSONObject jsonObj = JSON.parseObject(jsonParam);
           String eventCode = jsonObj.getString(Constants.EVENT_CODE_FILED);
           if(Constants.COMMIT_FILE_EVENT_CODE.equals(eventCode) || Constants.DEPLOY_FILE_EVENT_CODE.equals(eventCode)){
               //初始化client
               IAcsClient client = dataWorksOpenApiClient.createClient();
               try {
                   //當前事件參數信息
                   String messageId = jsonObj.getString("id");
                   JSONObject data = jsonObj.getObject("data", JSONObject.class);
                  // Long projectId = data.getLong("appId");
   
                   //初始化事件回調
                   CallbackExtensionRequest callbackExtensionRequest = new CallbackExtensionRequest();
                   callbackExtensionRequest.setMessageId(messageId);
                   callbackExtensionRequest.setExtensionCode(extensionParamProperties.getExtensionCode());
                   JSONObject eventBody = data.getJSONObject("eventBody");
                   Long dataRowSize = eventBody.getLong("dataRowSize");
                   //獲取擴展程序選項配置在項目空間下的配置
                   GetOptionValueForProjectRequest getOptionValueForProjectRequest = new GetOptionValueForProjectRequest();
                   //全局擴展點事件的配置信息所屬projectId默認為-1
                   getOptionValueForProjectRequest.setProjectId("-1");
                   getOptionValueForProjectRequest.setExtensionCode(extensionParamProperties.getExtensionCode());
                   GetOptionValueForProjectResponse getOptionValueForProjectResponse = client.getAcsResponse(getOptionValueForProjectRequest);
                   JSONObject jsonObject = JSON.parseObject(getOptionValueForProjectResponse.getOptionValue());
                   //這里需根據在DataWorks上實際設置格式來填寫
                   Long maxDataRowSize = jsonObject.getLong("dataRowSize");
                   //判斷代碼是否包含限制函數
                   if(dataRowSize > 1000){
   
                       callbackExtensionRequest.setCheckResult(EventCheckEnum.FAIL.getCode());
                       callbackExtensionRequest.setCheckMessage("下載的行數超過限制數");
                   }else{//成功回調
                       callbackExtensionRequest.setCheckResult(EventCheckEnum.OK.getCode());
                   }
                   //回調DataWorks
                   CallbackExtensionResponse acsResponse = client.getAcsResponse(callbackExtensionRequest);
                   //請求的唯一標識，用于后續錯誤排查使用
                   System.out.println("acsResponse:" + acsResponse.getRequestId());
               } catch (ClientException e) {
                   //請求的唯一標識，用于后續錯誤排查使用
                   System.out.println("RequestId:" + e.getRequestId());
                   //錯誤狀態碼
                   System.out.println("ErrCode:" + e.getErrCode());
                   //錯誤描述信息
                   System.out.println("ErrMsg:" + e.getErrMsg());
               }
           }else{
               System.out.println("未能過濾其他事件,請檢查配置步驟");
           }
       }
   }

3. 配置風險響應規則。

   進入安全中心 > 風險識別規則，為上述已發布上線的擴展程序配置風險響應，如需審批，則可添加已創建的審批流程。

   說明

   在響應配置時：

   • 如需實現“審批”，則需保證擴展程序在識別到用戶風險行為時callbackExtensionRequest.setCheckResult()返回“WARN”。

   • 如需實現“阻斷”，則callbackExtensionRequest.setCheckResult()應返回“FAIL”。

   

4. 開啟擴展程序。

1. 在數據開發、數據分析模塊單擊下載數據，將跳轉至數據下載頁面進行風險檢測。

2. 根據檢測結果進行后續處理。

   • 若檢測通過，則可繼續下載。

   

   • 若檢測不通過，則下載被阻斷，或告知用戶需申請權限。

     • 下載被阻斷。

       

     • 提示用戶申請權限。