ASP.Net Core中调用Databricks API执行SQL的高效等待方案问询

针对你的需求，以下是一套实用的高效交互方案，涵盖智能轮询、异步处理、流量控制等核心要点：

1. 基于命令状态的智能轮询策略 #

摒弃固定延迟等待，根据Databricks返回的命令状态动态调整轮询间隔，既减少无效调用，又能及时获取结果：

private async Task<CommandStatusResponse> PollCommandStatusAsync(string commandId, CancellationToken cancellationToken)
{
    int currentDelayMs = 1000; // 初始轮询间隔1秒
    const int maxDelayMs = 10000; // 最大间隔不超过10秒

    while (!cancellationToken.IsCancellationRequested)
    {
        // 调用Databricks状态接口
        var statusResp = await _databricksClient.GetCommandStatusAsync(commandId);

        switch (statusResp.Status)
        {
            // 终端状态，直接返回
            case "finished":
            case "error":
            case "cancelled":
                return statusResp;
            // 运行中，逐步延长间隔，避免高频调用
            case "running":
                currentDelayMs = Math.Min(currentDelayMs + 1000, maxDelayMs);
                break;
            // 排队中，间隔延长幅度更大
            case "pending":
                currentDelayMs = Math.Min(currentDelayMs + 2000, maxDelayMs);
                break;
        }

        await Task.Delay(currentDelayMs, cancellationToken);
    }

    throw new OperationCanceledException("命令轮询被取消");
}

2. 异步非阻塞的API接口实现 #

利用ASP.Net Core的异步特性，避免线程池阻塞，提升并发处理能力：

[HttpPost("execute-sql")]
public async Task<IActionResult> ExecuteSql([FromBody] SqlExecuteRequest request, CancellationToken cancellationToken)
{
    // 1. 提交SQL执行请求
    var executeResp = await _databricksClient.SubmitSqlCommandAsync(request.SqlContent, cancellationToken);
    if (!executeResp.IsSuccess)
    {
        return BadRequest(new { Message = executeResp.ErrorMsg });
    }

    // 2. 启动智能轮询
    var finalStatus = await PollCommandStatusAsync(executeResp.CommandId, cancellationToken);

    // 3. 处理并返回结果
    return finalStatus.Status switch
    {
        "finished" => Ok(finalStatus.QueryResults),
        _ => StatusCode(500, new { Message = $"执行失败: {finalStatus.ErrorMsg}" })
    };
}

3. 请求限流与队列缓冲 #

针对一定量的并发请求，通过限流和队列控制，避免压垮Databricks接口：

启用内置限流中间件 #

// Program.cs 配置限流
builder.Services.AddRateLimiter(options =>
{
    options.GlobalLimiter = PartitionedRateLimiter.Create<HttpContext, string>(context =>
        RateLimitPartition.GetFixedWindowLimiter(
            partitionKey: context.Request.Path,
            factory: _ => new FixedWindowRateLimiterOptions
            {
                PermitLimit = 20, // 允许同时处理20个请求
                Window = TimeSpan.FromMinutes(1),
                AutoReplenishment = true
            }));
});

// 启用限流中间件
app.UseRateLimiter();

后台队列异步处理（可选） #

如果请求峰值较高，可将请求放入内存队列，由后台服务异步处理，API直接返回任务ID：

// 注册队列与消费服务
builder.Services.AddSingleton<ConcurrentQueue<SqlExecutionTask>>();
builder.Services.AddHostedService<SqlTaskConsumer>();

// 消费服务实现
public class SqlTaskConsumer : BackgroundService
{
    private readonly ConcurrentQueue<SqlExecutionTask> _taskQueue;
    private readonly IServiceProvider _serviceProvider;

    public SqlTaskConsumer(ConcurrentQueue<SqlExecutionTask> taskQueue, IServiceProvider serviceProvider)
    {
        _taskQueue = taskQueue;
        _serviceProvider = serviceProvider;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            if (_taskQueue.TryDequeue(out var task))
            {
                using var scope = _serviceProvider.CreateScope();
                var client = scope.ServiceProvider.GetRequiredService<IDatabricksApiClient>();
                
                try
                {
                    var executeResp = await client.SubmitSqlCommandAsync(task.Sql, stoppingToken);
                    var statusResp = await PollCommandStatusAsync(executeResp.CommandId, stoppingToken);
                    
                    // 通知调用方结果（可通过SignalR、回调URL或数据库存储）
                    await task.CompleteCallback(statusResp);
                }
                catch (Exception ex)
                {
                    await task.ErrorCallback(ex);
                }
            }
            else
            {
                await Task.Delay(500, stoppingToken);
            }
        }
    }
}

对应的API接口：

[HttpPost("execute-sql-async")]
public IActionResult QueueSqlExecution([FromBody] SqlExecuteRequest request)
{
    var taskId = Guid.NewGuid().ToString();
    _taskQueue.Enqueue(new SqlExecutionTask
    {
        TaskId = taskId,
        Sql = request.SqlContent,
        CompleteCallback = async resp => { /* 处理结果通知逻辑 */ },
        ErrorCallback = async ex => { /* 处理错误通知逻辑 */ }
    });

    return Accepted(new { TaskId = taskId });
}

4. 可靠性增强：重试与资源复用 #

Polly重试策略处理API调用失败 #

// Program.cs 配置HttpClient重试
builder.Services.AddHttpClient<IDatabricksApiClient, DatabricksApiClient>()
    .AddPolicyHandler(Policy
        .Handle<HttpRequestException>()
        .OrResult<HttpResponseMessage>(r => !r.IsSuccessStatusCode)
        .WaitAndRetryAsync(3, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt))));

复用HttpClient避免资源耗尽 #

public class DatabricksApiClient : IDatabricksApiClient
{
    private readonly HttpClient _httpClient;

    public DatabricksApiClient(HttpClient httpClient)
    {
        _httpClient = httpClient;
        _httpClient.BaseAddress = new Uri("https://your-databricks-workspace/api/1.2/");
        _httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", "your-auth-token");
    }

    // 实现SubmitSqlCommandAsync和GetCommandStatusAsync方法
}

5. 可选：状态持久化 #

若需应对API重启场景，可将命令ID、请求信息、当前轮询状态存入数据库（如SQLite、SQL Server），重启后自动恢复未完成的轮询任务。

内容的提问来源于stack exchange，提问作者Work