Qishao.DingTalkStream.Core 2026.8.7

Qishao.DingTalkStream.Core

Qishao.DingTalkStream.Core 是钉钉 Stream 模式的基础库，负责注册订阅、建立长连接、接收推送、排队去重、并发调度消息处理器，以及生成推送响应数据。

安装

dotnet add package Qishao.DingTalkStream.Core

目标框架

• net8.0
• netstandard2.1

当前项目直接引用 Microsoft.Extensions.*。netstandard2.1 额外引用 System.Text.Json。

引用方式

当前仓库内使用项目引用：

<ProjectReference Include="..\Qishao.DingTalkStream.Core\Qishao.DingTalkStream.Core.csproj" />

快速开始

在 Generic Host 中注册 Stream 客户端、订阅和消息处理器：

using Qishao.DingTalkStream.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

var host = Host.CreateDefaultBuilder(args)
    .ConfigureServices((context, services) =>
    {
        services.AddDingtalkStream(options =>
        {
            options.ClientId = context.Configuration["ClientId"];
            options.ClientSecret = context.Configuration["ClientSecret"];
            options.AutoReplySystemMessage = true;
            options.MaxTaskCount = Environment.ProcessorCount;
        })
        .RegisterEventSubscription()
        .RegisterCardInstanceCallback()
        .AddMessageHandler<DefaultStreamMessageHandler>()
        .AddHostServices();
    })
    .Build();

await host.RunAsync();

消息处理器实现 IDingTalkStreamMessageHandler：

using Qishao.DingTalkStream.Core;

public sealed class DefaultStreamMessageHandler : IDingTalkStreamMessageHandler
{
    public async Task HandleMessage(MessageEventHanderArgs e)
    {
        if (e.Type != SubscriptionType.EVENT)
        {
            return;
        }

        var data = await DingTalkStreamUtilities.CreateReply_EventSuccess_MessageData("OK");
        var reply = await DingTalkStreamUtilities.CreateReplyMessage(e.Headers.MessageId, data);
        await e.Reply(reply);
    }
}

配置方式

推荐使用委托方式配置 ClientId 和 ClientSecret，因为当前 AddDingtalkStream(IConfiguration) 重载读取的是 ClientScript 键：

{
  "ClientId": "your-client-id",
  "ClientScript": "your-client-secret",
  "UA": "your-app/1.0.0",
  "AutoReplySystemMessage": "true",
  "Subscriptions": [
    {
      "Type": "EVENT",
      "Topic": "*"
    },
    {
      "Type": "CALLBACK",
      "Topic": "/v1.0/card/instances/callback"
    }
  ]
}

如果配置文件使用 ClientSecret，请使用 AddDingtalkStream(options => ...)。

注册 API

运行机制

1. DingTalkStreamClientWorker 延迟 1 秒启动。
2. Worker 创建 DingTalkStreamClient，并订阅客户端的 OnMessage 事件。
3. Client 使用 ClientId、ClientSecret 和订阅列表请求钉钉 Stream 网关连接信息。
4. Client 通过 ClientWebSocket 建立长连接并接收文本消息。
5. 开启 AutoReplySystemMessage 后，Client 自动处理 SYSTEM 类型的 ping 和 disconnect。
6. 非系统消息进入 Worker 队列，Worker 按 Headers.MessageId 做队列内和执行中去重。
7. Worker 每次从 DI scope 中解析一个 IDingTalkStreamMessageHandler 来处理消息。

主要模型

回复工具

DingTalkStreamUtilities 提供以下辅助方法：

并发和去重

• MaxTaskCount 控制同时处理消息的任务数，默认是 Environment.ProcessorCount。
• Worker 使用 ConcurrentQueue<MessageEventHanderArgs> 保存待处理消息。
• Worker 使用 ConcurrentDictionary<string, MessageEventHanderArgs> 记录执行中的消息。
• 相同 MessageId 已在队列中或正在执行时，新推送会被忽略。
• SingleExecuteTimeOut 只用于记录超时警告，不会取消处理器。
• RecentExecutionTimeCount 控制最近执行耗时的滑动窗口大小。
• MaxQueueCount 和 TimeInterval 当前只在 options 中定义，Worker 当前没有按这两个值限制队列或轮询。

注意点

• MessageEventHanderArgs、OnStoped、CreateReply_EventFaild_MessageData 等名称是当前公开 API，拼写保持兼容。
• AddDingtalkStream(IConfiguration) 使用 ClientScript 键读取密钥。
• Reply(byte[]) 应尽快调用。注释中说明 5 秒内未回复时，服务端可能重发。
• 事件订阅会在请求网关时合并为 EVENT + *。
• 非 EVENT 类型会按实际 Topic 写入订阅列表。

构建验证

dotnet build .\Qishao.DingTalkStream.Core\Qishao.DingTalkStream.Core.csproj --no-restore