尧图网页设计 
Bitwarden Server Event Integrations 架构解析双层 AMQP 消息流水线、重试机制与新集成扩展指南【免费下载链接】serverBitwarden infrastructure/backend (API, database, Docker, etc).项目地址: https://gitcode.com/GitHub_Trending/ser/server本文基于 Bitwarden 服务端仓库中的设计文档 src/Core/Dirt/EventIntegrations/README.md深入讲解 Bitwarden Server 事件集成Event Integrations子系统的整体架构如何借助 RabbitMQ / Azure Service Bus 的 fan-out 能力把组织审计事件广播到 Slack、Webhook、HEC、Datadog、Teams 等多种外部集成并通过“事件层 集成层”的双层 Exchange 设计实现独立重试、死信队列与模板化消息。读完本篇你将掌握该子系统的消息流转链路、重试与缓存机制并能够按照官方步骤独立完成一个新事件集成的接入与部署。一、设计目标事件集成子系统的核心目标是在不进行大量定制开发的前提下让新集成可以随时间轻松加入。具体包括四点原文档“Design goals”章节的完整继承Fan-out 即插即用借助 AMQPRabbitMQ 或 Azure Service Bus提供的扇出fan-out能力任何数量的新集成都可以挂接到现有事件系统上无需额外广播代码。只要向现有流水线添加一个新的 listener它就能获得一条独立的事件流。健壮的失败与重试处理通过下述双层 Exchange 方法在服务层面内建重试支持。新集成只需关注集成自身的业务逻辑与状态上报重试与延迟完全由消息系统托管。自托管Self-Hosted同等支持该能力不仅面向云端版本也对自托管实例开放。RabbitMQ 为自托管实例提供了一个轻量级的接入方式无需 Azure Service Bus 即可使用同一套健壮架构。组织管理员的灵活控制允许组织管理员决定哪些事件有重要意义、事件发送到哪里、以及消息中携带哪些数据。配置架构允许组织对特定集成的细节进行自定义见下文“集成与集成配置”一节。二、整体架构与入口IEventWriteService事件集成的入口是IEventWriteService。通过把EventIntegrationEventWriteService配置为EventWriteService所有发送到该服务的事件都会在 RabbitMQ 或 Azure Service Bus 的消息交换器exchange上广播。为屏蔽“发布到具体 AMQP 提供商”的细节EventIntegrationEventWriteService中注入了一个IEventIntegrationPublisher由它负责真正把事件发布到 RabbitMQ 或 Azure Service Bus。2.1 源码印证写服务的装配优先级从 EventIntegrationsServiceCollectionExtensions.cs 中的AddEventWriteServices扩展方法可以看到IEventWriteService的具体实现是按配置“择优”注册的优先级依次为Azure Service BusEventLogging.AzureServiceBus的连接串、事件 Topic、集成 Topic 三项齐全时注册EventIntegrationEventWriteService并以AzureServiceBusService作为IEventIntegrationPublisherRabbitMQEventLogging.RabbitMq的 HostName、Username、Password、EventExchangeName、IntegrationExchangeName 五项齐全时见IsRabbitMqEnabled同文件 #L541-L548注册EventIntegrationEventWriteServiceRabbitMqServiceAzure Queue Storage配置了Events.ConnectionString与Events.QueueName时注册AzureQueueEventWriteServiceRepository自托管SelfHosted为 true 时注册RepositoryEventWriteService事件直接落库Noop以上均未配置时注册不做任何事的NoopEventWriteService。这一装配顺序解释了文档中“云端存 Azure Tables、自托管存数据库”的行为差异来源没有配置消息总线时事件走仓储直写。2.2 发布侧实现EventIntegrationEventWriteService.cs 的实现非常薄CreateAsync把单个IEvent序列化为 JSONCreateManyAsync把事件数组序列化为 JSON取第一个事件的OrganizationId作为路由依据然后统一调用IEventIntegrationPublisher.PublishEventAsync。这正是文档所述“消息体是单个EventMessage或EventMessage数组的 JSON 表示”的发布端实现。三、双层 ExchangeTwo-tier Exchange当EventIntegrationEventWriteService发布消息时它发送到双层消息处理方案的第一层。每一层在 AMQP 栈中对应一个独立的 exchangeRabbitMQ 术语或 topicAzure Service Bus 术语。3.1 事件层Event Tier在第一层事件通过 fan-out 广播给一组 listener。消息体是单个EventMessage或EventMessage数组的 JSON 表示该层的 handler 负责处理每个事件或事件数组。目前有两个handlerEventRepositoryHandler负责事件的长期存储。它接收所有事件通过注入的IEventRepository存入数据库。这与事件集成关闭时的行为一致云端存 Azure Tables自托管存数据库。EventIntegrationHandler一个泛型通用类按集成的配置细节定制化职责是判断“该事件 / 该组织 / 该集成”是否存在配置、拉取该配置、并把事件详情解析进模板字符串。它使用注入的IOrganizationIntegrationConfigurationRepository按事件类型、组织、集成类型取出对应的配置与模板集合。这些配置决定了该集成是否应触发、发送所需细节、以及实际要发送的消息内容。其输出是一个新的IntegrationMessage携带与该集成交互所需的配置详情和已填入事件细节的待发消息并发布到消息总线的集成层。源码印证EventIntegrationHandler.cs 中HandleEventAsync的完整流程与文档描述一一对应按OrganizationId IntegrationType EventType从扩展缓存IFusionCache读取ListOrganizationIntegrationConfigurationDetails缓存未命中才回源到configurationRepository.GetManyByEventTypeOrganizationIdIntegrationTypeL126-L155若配置带Filters先反序列化为IntegrationFilterGroup并用IIntegrationFilterService求值为false则continue丢弃该事件L35-L44用IntegrationTemplateProcessor.ReplaceTokens渲染模板上下文按需从缓存补齐 Group / User / ActingUser / Organization 等实体默认 TTL 30 分钟组装IntegrationMessageTMessageId优先复用事件的IdempotencyId保证幂等调用eventIntegrationPublisher.PublishAsync发布到集成层L46-L65。3.2 集成层Integration Tier在集成层消息是IIntegrationMessage的 JSON 表示具体类型是泛型IntegrationMessageT的某个具体实现其中T即为该集成的配置详情。这些消息代表“把某个特定事件发送到某个特定集成”所需的全部细节包括重试与延迟处理。集成层的 handler 直接绑定到具体集成例如SlackIntegrationHandler、WebhookIntegrationHandler。它们接收IntegrationMessageT输出IntegrationHandlerResult告诉 listener 集成的结果成功/失败、是否可重试、以及最短延迟等。这种设计让它们可以在完全脱离 AMQP 与消息系统的情况下被单独单元测试。该层的 listener 负责收到新消息后调用 handler再根据结果采取行动——成功结果直接确认ack消息并结束失败则要么进入死信队列DLQ要么在正确的延迟后重新发布以重试。四、重试机制Retries引入集成层的目标之一就是简化并支撑针对单个事件集成的多次重试。例如某个下游服务短暂宕机时不希望某个 handler 阻塞其余队列的等待重试如果某个事件的 N 个集成中只有一个失败也不希望对其余集成全部重试更不希望重新查询配置。通过将IntegrationMessageT拆分为携带配置、消息体与重试细节的独立消息每个“事件/集成”组合都可以被独立处理、独立重试。当IntegrationHandlerResult.Success为false本次集成尝试失败时Retryable标志告诉 listener 该失败是暂时性的还是终局性的Retryable为false消息立即送入 DLQRetryable为truelistener 调用IntegrationMessage的ApplyRetry(DateTime)方法它同时负责递增RetryCount、按给定 DateTime 更新DelayUntilDate并叠加基于RetryCount的指数退避与随机抖动。listener 再比较RetryCount是否超过 Global Settings 中定义的MaxRetries超过则进 DLQ否则安排重试。4.1 源码印证退避算法与参数默认值IntegrationMessage.cs 中ApplyRetry的具体实现与文档完全吻合并给出了可复现的退避公式public void ApplyRetry(DateTime? handlerDelayUntilDate) { RetryCount; var baseTime handlerDelayUntilDate ?? DateTime.UtcNow; var backoffSeconds Math.Pow(2, RetryCount); // 指数退避2^n 秒 var jitterSeconds Random.Shared.Next(0, 3); // 0~2 秒随机抖动 DelayUntilDate baseTime.AddSeconds(backoffSeconds jitterSeconds); }即第 1、2、3 次重试的基础延迟分别为约 2s、4s、8s叠加 0~2s 抖动且 handler 可通过DelayUntilDate提供更晚的起始时间。相关默认值在 GlobalSettings.cs 中确认EventLogging.MaxRetries默认为3#L318EventLogging.RabbitMq.UseDelayPlugin默认为false#L369。ListenerConfiguration基类把MaxRetries直接绑定到globalSettings.EventLogging.MaxRetries见 ListenerConfiguration.cs。4.2 Azure Service Bus 的重试调度Azure Service Bus 在核心能力上原生支持消息定时重试被调度到某一具体时间点ASB 会扣留消息并在正确时间再发布。4.3 RabbitMQ 的重试选项仅自托管使用对 RabbitMQ仅自托管使用有两种选项由GlobalSettings.RabbitMqSettings中的useDelayPlugin标志决定。默认false表示使用“重试队列 时间检查”方案。选项 1延迟插件Delay Plugin对应 RabbitMQ 官方仓库中的rabbitmq-delayed-message-exchange插件可在 RabbitMQ 官方仓库搜索该名称。该插件在 RabbitMQ 中提供“延迟消息交换器”支持按特殊头header中指定的时长延迟一条消息。这使得方案可以完全不用重试队列直接依赖 delay exchange消息打上 header 后发布到 exchange由 exchange 负责在合适时间前扣留消息类似 ASB 的内建支持。该插件必须先安装并启用之后才能打开此选项因此默认关闭。选项 2重试队列 时间检查默认关闭延迟插件时消息被推入一个重试队列该队列有固定的时间长度后才把消息重新发布回主队列。消息从队列取出时检查DelayUntilDate是否已过期已过期正常处理集成并重试请求仍在未来把消息放回重试队列继续等待。虽然这消耗额外处理开销但在延迟插件未启用的情况下能更好地遵守延迟承诺。由于该方案仅用于自托管短延迟、少量重试下的开销可以忽略。源码印证RabbitMqIntegrationListenerService.cs 的ProcessReceivedMessageAsync完整实现了上述逻辑——先反序列化消息并检查DelayUntilDate是否仍在未来是则RepublishToRetryQueueAsync后 ack 返回否则调用 handler按result.Success / result.Retryable分支执行 ack、ApplyRetry后PublishToRetryAsync当RetryCount MaxRetries或PublishToDeadLetterAsync超出上限或不可重试。RabbitMqService.cs 中则依据_useDelayPlugin决定是在 header 中写入延迟时间、还是走 retry routing key。五、Listener / Handler 模式为了同时支持多个 AMQP 服务RabbitMQ 与 Azure Service Bus监听消息流的动作与响应消息的动作被解耦5.1 Listeners处理通信平台RabbitMQ / Azure Service Bus的细节每个平台、每个层级各有一个 listener即各有一个 event listener 和一个 integration listener负责消息平台的一切装配/拆除、订阅、消息确认ack等但自身不处理任何事件而是委托给与之配对的 handler可以配置多个实例独立运行各自拥有独立的 handler 与订阅/队列。5.2 Handlers每个队列/订阅集成层即每个集成对应一个 handler完全隔离于消息平台、对其一无所知因此可以跨通信平台自由复用负责事件处理的所有环节因其隔离与解耦具备高度可测试性。这种组合使得 ServiceCollectionExtensions 中的配置 可以把“当前消息平台的 listener 实例”与“任意数量的 handler”配对。AddEventIntegrationServices会为每个集成注册三样东西以AddAzureServiceBusIntegrationTConfig, TListenerConfig/AddRabbitMqIntegrationTConfig, TListenerConfig为例以ListenerConfiguration.RoutingKey为 key 的EventIntegrationHandlerTConfig键控单例事件层的*EventListenerService宿主服务集成层的*IntegrationListenerService宿主服务。目前代码中实际注册的集成 handler 包括 Slack、Webhook、Hec复用 Webhook handler 类型、Datadog、Teams 五个见 EventIntegrationsServiceCollectionExtensions.cs #L305-L316。六、Publishers 与 ServicesListeners以及EventIntegrationHandler通过IEventPublisher接口与消息系统交互其后是由 RabbitMQ 与 ASB 专属服务支撑的实现。把消息平台的大部分细节放在服务层可以在一处集中处理连接配置、绑定/创建特定队列等公共事务。IRabbitMqService与IAzureServiceBusService都实现了IEventPublisher接口因此也能直接处理所有消息发布功能。七、集成与集成配置Integrations Configurations组织可以为不同端点配置集成配置——每个 handler 映射到一个特定集成收到事件时检查对应配置。当前已有 Slack、Webhooks 与 HTTP Event CollectorHEC的集成/handler代码中还扩展了 Datadog 与 Teams。7.1OrganizationIntegration组织级启用某个集成的顶层对象包含适用于“该集成所有事件”的属性。例如Slack 把 token 存在Configuration中对每个事件都生效而把 channel id 存在OrganizationIntegrationConfiguration的Configuration中。token 适用于整个 Slack 集成但 channel 可以按事件类型不同而不同。各级具体存什么见下表。7.2OrganizationIntegrationConfiguration包含该集成针对每个EventType的配置Configuration包含事件级配置。此层级的任何属性都会覆盖OrganizationIntegration中的Configuration具体集成的例子见下表。Template包含模板字符串预期用实际事件内容填充。字符串中的 token 用#字符包裹例如 UserId 写作#UserId#。IntegrationTemplateProcessor负责把 token 替换为从给定EventMessage内省introspect出来的值。模板不强制任何结构——它既可以是发到 Slack 的自由文本也可以是发到 webhook 的 JSON body它只作为字符串存储和使用以最大化灵活性。7.3OrganizationIntegrationConfigurationDetails它是OrganizationIntegration与OrganizationIntegrationConfiguration二合一的合并对象合并内容告诉集成的 handler 调用外部服务所需的全部细节OrganizationIntegrationConfiguration优先于OrganizationIntegration——两者都存在的 key取OrganizationIntegrationConfiguration的值EventIntegrationHandler从数据库取出的正是一个OrganizationIntegrationConfigurationDetails数组用于决定在集成层发布什么。7.4 现有集成及其两级配置明细下表说明各集成如何配置、Configuration属性在两级OrganizationIntegration或OrganizationIntegrationConfiguration中分别存什么。OrganizationIntegration列中有效的OrganizationIntegrationStatus以粗体标出并给出每种状态下的存储示例。集成OrganizationIntegrationOrganizationIntegrationConfigurationCloudBillingSync不适用尚未使用不适用尚未使用Scim不适用尚未使用不适用尚未使用SlackInitiatednullCompleted{ Token: xoxb-token-from-slack }{ channelId: C123456 }Webhooknull或{ Scheme: Bearer, Token: AUTH-TOKEN, Uri: https://example.com }null或{ Scheme: Bearer, Token:AUTH-TOKEN, Uri: https://example.com }此层级定义的值优先Hec{ Scheme: Bearer, Token: AUTH-TOKEN, Uri: https://example.com }恒为nullDatadog{ ApiKey: TheKey12345, Uri: https://api.us5.datadoghq.com/api/v1/events}恒为nullTeamsInitiatednullIn Progress{ TenantID: tenant, Teams: [Id: team, DisplayName: MyTeam]}Completed{ TenantID: tenant, Teams: [Id: team, DisplayName: MyTeam], ServiceUrl:https://example.com, ChannelId: channel-1234}恒为null八、过滤Filtering除了上文所述的集成配置能力组织管理员还可以在OrganizationIntegrationConfiguration中添加可选的Filters。过滤器完全可选管理员可以把它做得简单或复杂。过滤器以 JSON 形式存库并被反序列化为IntegrationFilterGroup随后交给IntegrationFilterService求值为booltrue时集成按上述流程继续false时忽略该事件不路由到集成层。8.1IntegrationFilterGroup若干规则与其他子分组的逻辑 AND / OR 组合。属性说明AndOperator指示Rules与Groups中全部true还是任意false为真即可。该语义同时作用于内部分组与规则列表例如本组包含 Rule1、Rule2 以及 Group1、Group2 时trueRule1 Rule2 Group1 Group2falseRule1 \|\| Rule2 \|\| Group1 \|\| Group2RulesIntegrationFilterRule列表。可为 null 或空此时返回true。Groups嵌套的IntegrationFilterGroup列表。可为 null 或空此时返回true。8.2IntegrationFilterRule过滤框架的核心判断本条 EventMessage 的数据是否匹配过滤器所查找的数据。属性说明PropertyEventMessage上要评估的属性例如CollectionId。Operation属性与Value之间执行的比较。支持的操作•EqualsGuid等于Value•NotEqualsEquals的逻辑反向•InGuid在Value列表中•NotInIn的逻辑反向Value比较值。类型取决于Operation•Equals、NotEqualsGuid•In、NotInGuid列表源码印证EventIntegrationHandler.HandleEventAsync中对Filters字段的处理反序列化 EvaluateFilterGroup求值为false即continue就是这段描述的直接实现见 EventIntegrationHandler.cs #L35-L44。九、缓存Caching为降低数据库负载、提升性能事件集成使用自己独立的具名扩展缓存extended cache更多信息见 src/Core/Utilities/CACHING.md。没有缓存时每条进入的EventMessage都会触发一次数据库查询去取相关的OrganizationIntegrationConfigurationDetails。9.1EventIntegrationsCacheConstants该常量类让代码在操作扩展缓存时能以强类型引用大量缓存细节缓存名以及所有缓存 key 和 tag 都从EventIntegrationsCacheConstants程序化访问而不是散落的字符串字面量。例如EventIntegrationsCacheConstants.CacheName被用于缓存装配、键控服务、依赖注入等场景而不是在代码里写死字符串 EventIntegrations。9.2OrganizationIntegrationConfigurationDetails的缓存策略这是架构中被使用最频繁的部件之一任何带有组织的关联事件都需要检查配置判断是否要触发集成借助扩展缓存所有读操作在触及数据库前都先命中 L1 或 L2 缓存读取返回给定 key 的ListOrganizationIntegrationConfigurationDetails无匹配时返回空列表这些记录的 TTL 设得很高1 天。原因是管理端 API 每做任何变更都会通知缓存删除对应 key该删除会通过扩展缓存的 backplane 传播到事件监听代码缓存随即失效下次读取时拉取新值。因此高 TTL 是安全的——只在必要时才刷新。按集成打标签Tagging per integration缓存中的每条条目返回ListOrganizationIntegrationConfigurationDetails都会打上组织 id 与集成类型的标签这让管理员在集成层面做变更时可以一次性移除某组织某集成的全部配置详情。例如某组织的 webhook 配置了 5 个事件管理员改了集成层的 URL变更就必须被传播否则缓存会继续返回过期的 URL通过对每条条目打标签API 可以一次调用请求扩展缓存移除某个组织集成的全部条目缓存会以高性能方式处理这些条目的丢弃/刷新代码中有两处都知道标签机制且必须保持同步EventIntegrationHandler在拉取相关配置详情时必须使用该 tag——这样缓存从仓储成功加载时会带上 tag 存储该条目CreateOrganizationIntegrationCommand、UpdateOrganizationIntegrationCommand与DeleteOrganizationIntegrationCommand命令在管理员创建、更新、删除集成时必须使用该 tag 移除所有带 tag 的条目为保证两处对“如何打 tag”的认知一致它们都调用EventIntegrationsCacheConstants.BuildCacheTagForOrganizationIntegration构建 tag。源码印证EventIntegrationHandler.cs #L135-L152 中正是用BuildCacheTagForOrganizationIntegration(organizationId, integrationType)生成 tag并以FusionCacheEntryOptions(duration: DurationForOrganizationIntegrationConfigurationDetails)的高 TTL 选项执行cache.GetOrSetAsync与文档描述完全一致。9.3 模板属性缓存Template PropertiesIntegrationTemplateProcessor支持一些需要额外查询的属性。例如UserId直接来自EventMessage但UserName意味着要把 user id 映射到真实姓名的额外查询User含ActingUser、Group与Organization的属性通过扩展缓存缓存默认 TTL 30 分钟这些属性同时缓存在 L1内存与 L2Redis并在需要时自动刷新。十、构建一个新集成Building a New Integration以下是构建新集成所需的全部部件。为便于命名说明下文假设新集成名为 “Example”。完整示例可参考 Bitwarden 官方仓库中新增 Datadog 集成的 PR #6289可检索该 PR 号查看提交上下文。10.1 IntegrationType为IntegrationType枚举添加新集成的类型。10.2 Configuration Models配置模型配置模型决定OrganizationIntegration与OrganizationIntegrationConfiguration在数据库里存什么。Configuration列就是对应对象的序列化版本代表该集成与事件类型的配置细节ExampleIntegration整个集成的配置细节例如 Slack 的 token对该集成定义的每一个事件类型配置生效映射到OrganizationIntegration的Configuration中存储的 JSON 结构。ExampleIntegrationConfiguration可能逐事件变化的配置细节例如 Slack 的 channelId映射到OrganizationIntegrationConfiguration的Configuration中存储的 JSON 结构。ExampleIntegrationConfigurationDetailsIntegration 与 IntegrationConfiguration 的合并配置即OrganizationIntegrationConfigurationDetails中MergedConfiguration的反序列化结果。新增集成后应在上文“现有集成及其两级配置明细”表格中为新集成添加一行。10.3 Request Models请求模型在OrganizationIntegrationRequestModel.Validate的 switch 方法中新增分支——同时在OrganizationIntegrationRequestModelTests中添加测试在OrganizationIntegrationConfigurationRequestModel.IsValidForType的 switch 方法中新增分支——同时在OrganizationIntegrationConfigurationRequestModelTests中添加/更新测试。10.4 Response Model响应模型在OrganizationIntegrationResponseModel.Status的 switch 方法中新增分支——同时在OrganizationIntegrationResponseModelTests中添加/更新测试。10.5 Integration Handler集成处理器例如ExampleIntegrationHandler这里存放执行集成动作的实际代码即发出 HTTP 请求等Handler 接收IntegrationMessageT其中T是上面定义的ExampleIntegrationConfigurationDetails包含 Configuration 与已渲染好待发送的模板消息Handler 返回IntegrationHandlerResult携带请求结果细节——成功/失败、是否可重试、应延迟到何时等Handler 的职责范围仅仅是“执行集成并上报结果”。其余一切重试次数、何时重试、失败如何处理都由 Listener 负责。10.6 GlobalSettings全局设置RabbitMQ添加该集成的队列名。它们通常带有默认值这样首次被代码访问时 RabbitMQ 会自动创建它们ExampleEventQueueNameExampleIntegrationQueueNameExampleIntegrationRetryQueueNameAzure Service Bus添加 ASB 使用的订阅名。与 RabbitMQ 类似也提供默认值避免必须写入 secrets 才能配置同时允许覆盖。但是与 RabbitMQ 不同这些订阅必须在代码访问它们之前就已存在不会按需自动创建。参见下文“部署新集成”。ExampleEventSubscriptionNameExampleIntegrationSubscriptionNameService Bus Emulator 本地配置为在本地创建 ASB 资源还需要更新 dev/servicebusemulator_config.json 加入新订阅在现有事件 topicevent-logging下为新集成的事件层添加订阅events-example-subscription在现有集成 topicevent-integrations下为集成层消息添加新订阅integration-example-subscription从其他集成层订阅复制 correlation filter。它应基于IntegrationType.ToRoutingKey过滤本例即example。此处添加的名称必须与 secrets 中提供的值或 Global Settings 中给出的默认值一致。必须就位并重启本地 ASB emulator之后才能使用任何本地访问 ASB 资源的代码。10.7 ListenerConfiguration监听器配置新集成需要一个ListenerConfiguration的子类且该类还需实现IIntegrationListenerConfiguration。这个类提供访问前文在GlobalSettings中配置的 RabbitMQ 队列与 ASB 订阅的途径。新的 listener 配置将用于对 listener 进行类型化并提供访问该集成所需配置的手段。10.8 ServiceCollectionExtensions服务装配在ServiceCollectionExtensions中把上述所有部件串联起来在每一消息层启动带 handler 的 listener。所有事件集成装配的核心方法是AddEventIntegrationServices。两个“添加 listener”的方法都会调用它从而保证跨消息平台的公共依赖与集成只有唯一装配点。例如SlackIntegrationHandler需要SlackService所以AddEventIntegrationServices里有AddSlackService的调用webhook 的具名 HttpClient 定义同理。在AddEventIntegrationServices中创建 handler 的单例services.TryAddSingletonIIntegrationHandlerExampleIntegrationConfigurationDetails, ExampleIntegrationHandler();创建 listener 配置var exampleConfiguration new ExampleListenerConfiguration(globalSettings);把集成加入 RabbitMQ 与 ASB 各自的声明services.AddRabbitMqIntegrationExampleIntegrationConfigurationDetails, ExampleListenerConfiguration(exampleConfiguration);以及services.AddAzureServiceBusIntegrationExampleIntegrationConfigurationDetails, ExampleListenerConfiguration(exampleConfiguration);以上三步与当前仓库中 AddEventIntegrationServices 的实际写法注册 handler 单例 → 构造各 ListenerConfiguration → 按 ASB/RabbitMQ 分别调用Add*Integration泛型扩展完全对应。十一、部署新集成Deploying a New Integration11.1 RabbitMQRabbitMQ 会在队列和交换器首次被代码访问时动态创建它们。因此部署新集成时无需手工创建队列。当然也可以提前创建和配置但不是必须。注意一旦创建完成若需要修改任何配置队列或交换器必须删除后重建。11.2 Azure Service Bus与 RabbitMQ 相反ASB 资源必须在代码访问之前分配不会按需创建。这意味着新集成所需的任何订阅都必须在部署该代码之前在 ASB 中创建好。上文在 Global Settings 与servicebusemulator_config.json中定义的两个订阅需要在部署代码前通过 Azure 门户或 CLI 为对应环境创建ExampleEventSubscriptionName这是从主事件 topic 扇出fan-out的订阅因此它一经声明就会开始接收所有事件这可能在“集成专属 handler 声明并部署”之前形成积压backlog一种规避策略是以假过滤器例如1 0创建该订阅订阅被创建但过滤器确保没有任何消息真正落入该订阅可以部署引用该订阅的代码因为订阅合法存在只是为空代码就位、准备让新集成开始接收消息时移除过滤器即可让订阅恢复接收所有 fan-out 消息。ExampleIntegrationSubscriptionName该订阅必须在新集成代码部署前创建但它不是 fan-out而是基于IntegrationType.ToRoutingKey的过滤器因此在组织拥有激活的配置之前它不会开始接收消息。这意味着提前声明它不会造成积压风险。十二、小结关键路径速查关注点关键类型/文件事件发布入口IEventWriteService / EventIntegrationEventWriteService.cs平台选择与装配EventIntegrationsServiceCollectionExtensions.cs事件层处理器EventIntegrationHandler.cs、EventRepositoryHandler.cs集成层重试RabbitMqIntegrationListenerService.cs、AzureServiceBusIntegrationListenerService.cs消息与退避IntegrationMessage.cs重试/延迟参数GlobalSettings.csEventLogging.MaxRetries、EventLogging.RabbitMq.UseDelayPlugin集成配置管理命令CreateOrganizationIntegrationCommand.cs、UpdateOrganizationIntegrationCommand.cs、DeleteOrganizationIntegrationCommand.cs本地 ASB 资源dev/servicebusemulator_config.json需要说明的适用前提本架构中的 Azure Service Bus 路径对应云端部署RabbitMQ 路径主要面向自托管实例两者的资源创建语义按需创建 vs 预先创建是部署新集成时最需要注意的差异。所有队列/订阅默认值均可通过 Global Settings 或 secrets 覆盖但修改已创建的 RabbitMQ 队列/交换器配置需要先删除再重建。【免费下载链接】serverBitwarden infrastructure/backend (API, database, Docker, etc).项目地址: https://gitcode.com/GitHub_Trending/ser/server创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Bitwarden Server Event Integrations 架构解析：双层 AMQP 消息流水线、重试机制与新集成扩展指南
Bitwarden Server Event Integrations 架构解析双层 AMQP 消息流水线、重试机制与新集成扩展指南【免费下载链接】serverBitwarden infrastructure/backend (API, database, Docker, etc).项目地址: https://gitcode.com/GitHub_Trending/ser/server本文基于 Bitwarden 服务端仓库中的设计文档 src/Core/Dirt/EventIntegrations/README.md深入讲解 Bitwarden Server 事件集成Event Integrations子系统的整体架构如何借助 RabbitMQ / Azure Service Bus 的 fan-out 能力把组织审计事件广播到 Slack、Webhook、HEC、Datadog、Teams 等多种外部集成并通过“事件层 集成层”的双层 Exchange 设计实现独立重试、死信队列与模板化消息。读完本篇你将掌握该子系统的消息流转链路、重试与缓存机制并能够按照官方步骤独立完成一个新事件集成的接入与部署。一、设计目标事件集成子系统的核心目标是在不进行大量定制开发的前提下让新集成可以随时间轻松加入。具体包括四点原文档“Design goals”章节的完整继承Fan-out 即插即用借助 AMQPRabbitMQ 或 Azure Service Bus提供的扇出fan-out能力任何数量的新集成都可以挂接到现有事件系统上无需额外广播代码。只要向现有流水线添加一个新的 listener它就能获得一条独立的事件流。健壮的失败与重试处理通过下述双层 Exchange 方法在服务层面内建重试支持。新集成只需关注集成自身的业务逻辑与状态上报重试与延迟完全由消息系统托管。自托管Self-Hosted同等支持该能力不仅面向云端版本也对自托管实例开放。RabbitMQ 为自托管实例提供了一个轻量级的接入方式无需 Azure Service Bus 即可使用同一套健壮架构。组织管理员的灵活控制允许组织管理员决定哪些事件有重要意义、事件发送到哪里、以及消息中携带哪些数据。配置架构允许组织对特定集成的细节进行自定义见下文“集成与集成配置”一节。二、整体架构与入口IEventWriteService事件集成的入口是IEventWriteService。通过把EventIntegrationEventWriteService配置为EventWriteService所有发送到该服务的事件都会在 RabbitMQ 或 Azure Service Bus 的消息交换器exchange上广播。为屏蔽“发布到具体 AMQP 提供商”的细节EventIntegrationEventWriteService中注入了一个IEventIntegrationPublisher由它负责真正把事件发布到 RabbitMQ 或 Azure Service Bus。2.1 源码印证写服务的装配优先级从 EventIntegrationsServiceCollectionExtensions.cs 中的AddEventWriteServices扩展方法可以看到IEventWriteService的具体实现是按配置“择优”注册的优先级依次为Azure Service BusEventLogging.AzureServiceBus的连接串、事件 Topic、集成 Topic 三项齐全时注册EventIntegrationEventWriteService并以AzureServiceBusService作为IEventIntegrationPublisherRabbitMQEventLogging.RabbitMq的 HostName、Username、Password、EventExchangeName、IntegrationExchangeName 五项齐全时见IsRabbitMqEnabled同文件 #L541-L548注册EventIntegrationEventWriteServiceRabbitMqServiceAzure Queue Storage配置了Events.ConnectionString与Events.QueueName时注册AzureQueueEventWriteServiceRepository自托管SelfHosted为 true 时注册RepositoryEventWriteService事件直接落库Noop以上均未配置时注册不做任何事的NoopEventWriteService。这一装配顺序解释了文档中“云端存 Azure Tables、自托管存数据库”的行为差异来源没有配置消息总线时事件走仓储直写。2.2 发布侧实现EventIntegrationEventWriteService.cs 的实现非常薄CreateAsync把单个IEvent序列化为 JSONCreateManyAsync把事件数组序列化为 JSON取第一个事件的OrganizationId作为路由依据然后统一调用IEventIntegrationPublisher.PublishEventAsync。这正是文档所述“消息体是单个EventMessage或EventMessage数组的 JSON 表示”的发布端实现。三、双层 ExchangeTwo-tier Exchange当EventIntegrationEventWriteService发布消息时它发送到双层消息处理方案的第一层。每一层在 AMQP 栈中对应一个独立的 exchangeRabbitMQ 术语或 topicAzure Service Bus 术语。3.1 事件层Event Tier在第一层事件通过 fan-out 广播给一组 listener。消息体是单个EventMessage或EventMessage数组的 JSON 表示该层的 handler 负责处理每个事件或事件数组。目前有两个handlerEventRepositoryHandler负责事件的长期存储。它接收所有事件通过注入的IEventRepository存入数据库。这与事件集成关闭时的行为一致云端存 Azure Tables自托管存数据库。EventIntegrationHandler一个泛型通用类按集成的配置细节定制化职责是判断“该事件 / 该组织 / 该集成”是否存在配置、拉取该配置、并把事件详情解析进模板字符串。它使用注入的IOrganizationIntegrationConfigurationRepository按事件类型、组织、集成类型取出对应的配置与模板集合。这些配置决定了该集成是否应触发、发送所需细节、以及实际要发送的消息内容。其输出是一个新的IntegrationMessage携带与该集成交互所需的配置详情和已填入事件细节的待发消息并发布到消息总线的集成层。源码印证EventIntegrationHandler.cs 中HandleEventAsync的完整流程与文档描述一一对应按OrganizationId IntegrationType EventType从扩展缓存IFusionCache读取ListOrganizationIntegrationConfigurationDetails缓存未命中才回源到configurationRepository.GetManyByEventTypeOrganizationIdIntegrationTypeL126-L155若配置带Filters先反序列化为IntegrationFilterGroup并用IIntegrationFilterService求值为false则continue丢弃该事件L35-L44用IntegrationTemplateProcessor.ReplaceTokens渲染模板上下文按需从缓存补齐 Group / User / ActingUser / Organization 等实体默认 TTL 30 分钟组装IntegrationMessageTMessageId优先复用事件的IdempotencyId保证幂等调用eventIntegrationPublisher.PublishAsync发布到集成层L46-L65。3.2 集成层Integration Tier在集成层消息是IIntegrationMessage的 JSON 表示具体类型是泛型IntegrationMessageT的某个具体实现其中T即为该集成的配置详情。这些消息代表“把某个特定事件发送到某个特定集成”所需的全部细节包括重试与延迟处理。集成层的 handler 直接绑定到具体集成例如SlackIntegrationHandler、WebhookIntegrationHandler。它们接收IntegrationMessageT输出IntegrationHandlerResult告诉 listener 集成的结果成功/失败、是否可重试、以及最短延迟等。这种设计让它们可以在完全脱离 AMQP 与消息系统的情况下被单独单元测试。该层的 listener 负责收到新消息后调用 handler再根据结果采取行动——成功结果直接确认ack消息并结束失败则要么进入死信队列DLQ要么在正确的延迟后重新发布以重试。四、重试机制Retries引入集成层的目标之一就是简化并支撑针对单个事件集成的多次重试。例如某个下游服务短暂宕机时不希望某个 handler 阻塞其余队列的等待重试如果某个事件的 N 个集成中只有一个失败也不希望对其余集成全部重试更不希望重新查询配置。通过将IntegrationMessageT拆分为携带配置、消息体与重试细节的独立消息每个“事件/集成”组合都可以被独立处理、独立重试。当IntegrationHandlerResult.Success为false本次集成尝试失败时Retryable标志告诉 listener 该失败是暂时性的还是终局性的Retryable为false消息立即送入 DLQRetryable为truelistener 调用IntegrationMessage的ApplyRetry(DateTime)方法它同时负责递增RetryCount、按给定 DateTime 更新DelayUntilDate并叠加基于RetryCount的指数退避与随机抖动。listener 再比较RetryCount是否超过 Global Settings 中定义的MaxRetries超过则进 DLQ否则安排重试。4.1 源码印证退避算法与参数默认值IntegrationMessage.cs 中ApplyRetry的具体实现与文档完全吻合并给出了可复现的退避公式public void ApplyRetry(DateTime? handlerDelayUntilDate) { RetryCount; var baseTime handlerDelayUntilDate ?? DateTime.UtcNow; var backoffSeconds Math.Pow(2, RetryCount); // 指数退避2^n 秒 var jitterSeconds Random.Shared.Next(0, 3); // 0~2 秒随机抖动 DelayUntilDate baseTime.AddSeconds(backoffSeconds jitterSeconds); }即第 1、2、3 次重试的基础延迟分别为约 2s、4s、8s叠加 0~2s 抖动且 handler 可通过DelayUntilDate提供更晚的起始时间。相关默认值在 GlobalSettings.cs 中确认EventLogging.MaxRetries默认为3#L318EventLogging.RabbitMq.UseDelayPlugin默认为false#L369。ListenerConfiguration基类把MaxRetries直接绑定到globalSettings.EventLogging.MaxRetries见 ListenerConfiguration.cs。4.2 Azure Service Bus 的重试调度Azure Service Bus 在核心能力上原生支持消息定时重试被调度到某一具体时间点ASB 会扣留消息并在正确时间再发布。4.3 RabbitMQ 的重试选项仅自托管使用对 RabbitMQ仅自托管使用有两种选项由GlobalSettings.RabbitMqSettings中的useDelayPlugin标志决定。默认false表示使用“重试队列 时间检查”方案。选项 1延迟插件Delay Plugin对应 RabbitMQ 官方仓库中的rabbitmq-delayed-message-exchange插件可在 RabbitMQ 官方仓库搜索该名称。该插件在 RabbitMQ 中提供“延迟消息交换器”支持按特殊头header中指定的时长延迟一条消息。这使得方案可以完全不用重试队列直接依赖 delay exchange消息打上 header 后发布到 exchange由 exchange 负责在合适时间前扣留消息类似 ASB 的内建支持。该插件必须先安装并启用之后才能打开此选项因此默认关闭。选项 2重试队列 时间检查默认关闭延迟插件时消息被推入一个重试队列该队列有固定的时间长度后才把消息重新发布回主队列。消息从队列取出时检查DelayUntilDate是否已过期已过期正常处理集成并重试请求仍在未来把消息放回重试队列继续等待。虽然这消耗额外处理开销但在延迟插件未启用的情况下能更好地遵守延迟承诺。由于该方案仅用于自托管短延迟、少量重试下的开销可以忽略。源码印证RabbitMqIntegrationListenerService.cs 的ProcessReceivedMessageAsync完整实现了上述逻辑——先反序列化消息并检查DelayUntilDate是否仍在未来是则RepublishToRetryQueueAsync后 ack 返回否则调用 handler按result.Success / result.Retryable分支执行 ack、ApplyRetry后PublishToRetryAsync当RetryCount MaxRetries或PublishToDeadLetterAsync超出上限或不可重试。RabbitMqService.cs 中则依据_useDelayPlugin决定是在 header 中写入延迟时间、还是走 retry routing key。五、Listener / Handler 模式为了同时支持多个 AMQP 服务RabbitMQ 与 Azure Service Bus监听消息流的动作与响应消息的动作被解耦5.1 Listeners处理通信平台RabbitMQ / Azure Service Bus的细节每个平台、每个层级各有一个 listener即各有一个 event listener 和一个 integration listener负责消息平台的一切装配/拆除、订阅、消息确认ack等但自身不处理任何事件而是委托给与之配对的 handler可以配置多个实例独立运行各自拥有独立的 handler 与订阅/队列。5.2 Handlers每个队列/订阅集成层即每个集成对应一个 handler完全隔离于消息平台、对其一无所知因此可以跨通信平台自由复用负责事件处理的所有环节因其隔离与解耦具备高度可测试性。这种组合使得 ServiceCollectionExtensions 中的配置 可以把“当前消息平台的 listener 实例”与“任意数量的 handler”配对。AddEventIntegrationServices会为每个集成注册三样东西以AddAzureServiceBusIntegrationTConfig, TListenerConfig/AddRabbitMqIntegrationTConfig, TListenerConfig为例以ListenerConfiguration.RoutingKey为 key 的EventIntegrationHandlerTConfig键控单例事件层的*EventListenerService宿主服务集成层的*IntegrationListenerService宿主服务。目前代码中实际注册的集成 handler 包括 Slack、Webhook、Hec复用 Webhook handler 类型、Datadog、Teams 五个见 EventIntegrationsServiceCollectionExtensions.cs #L305-L316。六、Publishers 与 ServicesListeners以及EventIntegrationHandler通过IEventPublisher接口与消息系统交互其后是由 RabbitMQ 与 ASB 专属服务支撑的实现。把消息平台的大部分细节放在服务层可以在一处集中处理连接配置、绑定/创建特定队列等公共事务。IRabbitMqService与IAzureServiceBusService都实现了IEventPublisher接口因此也能直接处理所有消息发布功能。七、集成与集成配置Integrations Configurations组织可以为不同端点配置集成配置——每个 handler 映射到一个特定集成收到事件时检查对应配置。当前已有 Slack、Webhooks 与 HTTP Event CollectorHEC的集成/handler代码中还扩展了 Datadog 与 Teams。7.1OrganizationIntegration组织级启用某个集成的顶层对象包含适用于“该集成所有事件”的属性。例如Slack 把 token 存在Configuration中对每个事件都生效而把 channel id 存在OrganizationIntegrationConfiguration的Configuration中。token 适用于整个 Slack 集成但 channel 可以按事件类型不同而不同。各级具体存什么见下表。7.2OrganizationIntegrationConfiguration包含该集成针对每个EventType的配置Configuration包含事件级配置。此层级的任何属性都会覆盖OrganizationIntegration中的Configuration具体集成的例子见下表。Template包含模板字符串预期用实际事件内容填充。字符串中的 token 用#字符包裹例如 UserId 写作#UserId#。IntegrationTemplateProcessor负责把 token 替换为从给定EventMessage内省introspect出来的值。模板不强制任何结构——它既可以是发到 Slack 的自由文本也可以是发到 webhook 的 JSON body它只作为字符串存储和使用以最大化灵活性。7.3OrganizationIntegrationConfigurationDetails它是OrganizationIntegration与OrganizationIntegrationConfiguration二合一的合并对象合并内容告诉集成的 handler 调用外部服务所需的全部细节OrganizationIntegrationConfiguration优先于OrganizationIntegration——两者都存在的 key取OrganizationIntegrationConfiguration的值EventIntegrationHandler从数据库取出的正是一个OrganizationIntegrationConfigurationDetails数组用于决定在集成层发布什么。7.4 现有集成及其两级配置明细下表说明各集成如何配置、Configuration属性在两级OrganizationIntegration或OrganizationIntegrationConfiguration中分别存什么。OrganizationIntegration列中有效的OrganizationIntegrationStatus以粗体标出并给出每种状态下的存储示例。集成OrganizationIntegrationOrganizationIntegrationConfigurationCloudBillingSync不适用尚未使用不适用尚未使用Scim不适用尚未使用不适用尚未使用SlackInitiatednullCompleted{ Token: xoxb-token-from-slack }{ channelId: C123456 }Webhooknull或{ Scheme: Bearer, Token: AUTH-TOKEN, Uri: https://example.com }null或{ Scheme: Bearer, Token:AUTH-TOKEN, Uri: https://example.com }此层级定义的值优先Hec{ Scheme: Bearer, Token: AUTH-TOKEN, Uri: https://example.com }恒为nullDatadog{ ApiKey: TheKey12345, Uri: https://api.us5.datadoghq.com/api/v1/events}恒为nullTeamsInitiatednullIn Progress{ TenantID: tenant, Teams: [Id: team, DisplayName: MyTeam]}Completed{ TenantID: tenant, Teams: [Id: team, DisplayName: MyTeam], ServiceUrl:https://example.com, ChannelId: channel-1234}恒为null八、过滤Filtering除了上文所述的集成配置能力组织管理员还可以在OrganizationIntegrationConfiguration中添加可选的Filters。过滤器完全可选管理员可以把它做得简单或复杂。过滤器以 JSON 形式存库并被反序列化为IntegrationFilterGroup随后交给IntegrationFilterService求值为booltrue时集成按上述流程继续false时忽略该事件不路由到集成层。8.1IntegrationFilterGroup若干规则与其他子分组的逻辑 AND / OR 组合。属性说明AndOperator指示Rules与Groups中全部true还是任意false为真即可。该语义同时作用于内部分组与规则列表例如本组包含 Rule1、Rule2 以及 Group1、Group2 时trueRule1 Rule2 Group1 Group2falseRule1 \|\| Rule2 \|\| Group1 \|\| Group2RulesIntegrationFilterRule列表。可为 null 或空此时返回true。Groups嵌套的IntegrationFilterGroup列表。可为 null 或空此时返回true。8.2IntegrationFilterRule过滤框架的核心判断本条 EventMessage 的数据是否匹配过滤器所查找的数据。属性说明PropertyEventMessage上要评估的属性例如CollectionId。Operation属性与Value之间执行的比较。支持的操作•EqualsGuid等于Value•NotEqualsEquals的逻辑反向•InGuid在Value列表中•NotInIn的逻辑反向Value比较值。类型取决于Operation•Equals、NotEqualsGuid•In、NotInGuid列表源码印证EventIntegrationHandler.HandleEventAsync中对Filters字段的处理反序列化 EvaluateFilterGroup求值为false即continue就是这段描述的直接实现见 EventIntegrationHandler.cs #L35-L44。九、缓存Caching为降低数据库负载、提升性能事件集成使用自己独立的具名扩展缓存extended cache更多信息见 src/Core/Utilities/CACHING.md。没有缓存时每条进入的EventMessage都会触发一次数据库查询去取相关的OrganizationIntegrationConfigurationDetails。9.1EventIntegrationsCacheConstants该常量类让代码在操作扩展缓存时能以强类型引用大量缓存细节缓存名以及所有缓存 key 和 tag 都从EventIntegrationsCacheConstants程序化访问而不是散落的字符串字面量。例如EventIntegrationsCacheConstants.CacheName被用于缓存装配、键控服务、依赖注入等场景而不是在代码里写死字符串 EventIntegrations。9.2OrganizationIntegrationConfigurationDetails的缓存策略这是架构中被使用最频繁的部件之一任何带有组织的关联事件都需要检查配置判断是否要触发集成借助扩展缓存所有读操作在触及数据库前都先命中 L1 或 L2 缓存读取返回给定 key 的ListOrganizationIntegrationConfigurationDetails无匹配时返回空列表这些记录的 TTL 设得很高1 天。原因是管理端 API 每做任何变更都会通知缓存删除对应 key该删除会通过扩展缓存的 backplane 传播到事件监听代码缓存随即失效下次读取时拉取新值。因此高 TTL 是安全的——只在必要时才刷新。按集成打标签Tagging per integration缓存中的每条条目返回ListOrganizationIntegrationConfigurationDetails都会打上组织 id 与集成类型的标签这让管理员在集成层面做变更时可以一次性移除某组织某集成的全部配置详情。例如某组织的 webhook 配置了 5 个事件管理员改了集成层的 URL变更就必须被传播否则缓存会继续返回过期的 URL通过对每条条目打标签API 可以一次调用请求扩展缓存移除某个组织集成的全部条目缓存会以高性能方式处理这些条目的丢弃/刷新代码中有两处都知道标签机制且必须保持同步EventIntegrationHandler在拉取相关配置详情时必须使用该 tag——这样缓存从仓储成功加载时会带上 tag 存储该条目CreateOrganizationIntegrationCommand、UpdateOrganizationIntegrationCommand与DeleteOrganizationIntegrationCommand命令在管理员创建、更新、删除集成时必须使用该 tag 移除所有带 tag 的条目为保证两处对“如何打 tag”的认知一致它们都调用EventIntegrationsCacheConstants.BuildCacheTagForOrganizationIntegration构建 tag。源码印证EventIntegrationHandler.cs #L135-L152 中正是用BuildCacheTagForOrganizationIntegration(organizationId, integrationType)生成 tag并以FusionCacheEntryOptions(duration: DurationForOrganizationIntegrationConfigurationDetails)的高 TTL 选项执行cache.GetOrSetAsync与文档描述完全一致。9.3 模板属性缓存Template PropertiesIntegrationTemplateProcessor支持一些需要额外查询的属性。例如UserId直接来自EventMessage但UserName意味着要把 user id 映射到真实姓名的额外查询User含ActingUser、Group与Organization的属性通过扩展缓存缓存默认 TTL 30 分钟这些属性同时缓存在 L1内存与 L2Redis并在需要时自动刷新。十、构建一个新集成Building a New Integration以下是构建新集成所需的全部部件。为便于命名说明下文假设新集成名为 “Example”。完整示例可参考 Bitwarden 官方仓库中新增 Datadog 集成的 PR #6289可检索该 PR 号查看提交上下文。10.1 IntegrationType为IntegrationType枚举添加新集成的类型。10.2 Configuration Models配置模型配置模型决定OrganizationIntegration与OrganizationIntegrationConfiguration在数据库里存什么。Configuration列就是对应对象的序列化版本代表该集成与事件类型的配置细节ExampleIntegration整个集成的配置细节例如 Slack 的 token对该集成定义的每一个事件类型配置生效映射到OrganizationIntegration的Configuration中存储的 JSON 结构。ExampleIntegrationConfiguration可能逐事件变化的配置细节例如 Slack 的 channelId映射到OrganizationIntegrationConfiguration的Configuration中存储的 JSON 结构。ExampleIntegrationConfigurationDetailsIntegration 与 IntegrationConfiguration 的合并配置即OrganizationIntegrationConfigurationDetails中MergedConfiguration的反序列化结果。新增集成后应在上文“现有集成及其两级配置明细”表格中为新集成添加一行。10.3 Request Models请求模型在OrganizationIntegrationRequestModel.Validate的 switch 方法中新增分支——同时在OrganizationIntegrationRequestModelTests中添加测试在OrganizationIntegrationConfigurationRequestModel.IsValidForType的 switch 方法中新增分支——同时在OrganizationIntegrationConfigurationRequestModelTests中添加/更新测试。10.4 Response Model响应模型在OrganizationIntegrationResponseModel.Status的 switch 方法中新增分支——同时在OrganizationIntegrationResponseModelTests中添加/更新测试。10.5 Integration Handler集成处理器例如ExampleIntegrationHandler这里存放执行集成动作的实际代码即发出 HTTP 请求等Handler 接收IntegrationMessageT其中T是上面定义的ExampleIntegrationConfigurationDetails包含 Configuration 与已渲染好待发送的模板消息Handler 返回IntegrationHandlerResult携带请求结果细节——成功/失败、是否可重试、应延迟到何时等Handler 的职责范围仅仅是“执行集成并上报结果”。其余一切重试次数、何时重试、失败如何处理都由 Listener 负责。10.6 GlobalSettings全局设置RabbitMQ添加该集成的队列名。它们通常带有默认值这样首次被代码访问时 RabbitMQ 会自动创建它们ExampleEventQueueNameExampleIntegrationQueueNameExampleIntegrationRetryQueueNameAzure Service Bus添加 ASB 使用的订阅名。与 RabbitMQ 类似也提供默认值避免必须写入 secrets 才能配置同时允许覆盖。但是与 RabbitMQ 不同这些订阅必须在代码访问它们之前就已存在不会按需自动创建。参见下文“部署新集成”。ExampleEventSubscriptionNameExampleIntegrationSubscriptionNameService Bus Emulator 本地配置为在本地创建 ASB 资源还需要更新 dev/servicebusemulator_config.json 加入新订阅在现有事件 topicevent-logging下为新集成的事件层添加订阅events-example-subscription在现有集成 topicevent-integrations下为集成层消息添加新订阅integration-example-subscription从其他集成层订阅复制 correlation filter。它应基于IntegrationType.ToRoutingKey过滤本例即example。此处添加的名称必须与 secrets 中提供的值或 Global Settings 中给出的默认值一致。必须就位并重启本地 ASB emulator之后才能使用任何本地访问 ASB 资源的代码。10.7 ListenerConfiguration监听器配置新集成需要一个ListenerConfiguration的子类且该类还需实现IIntegrationListenerConfiguration。这个类提供访问前文在GlobalSettings中配置的 RabbitMQ 队列与 ASB 订阅的途径。新的 listener 配置将用于对 listener 进行类型化并提供访问该集成所需配置的手段。10.8 ServiceCollectionExtensions服务装配在ServiceCollectionExtensions中把上述所有部件串联起来在每一消息层启动带 handler 的 listener。所有事件集成装配的核心方法是AddEventIntegrationServices。两个“添加 listener”的方法都会调用它从而保证跨消息平台的公共依赖与集成只有唯一装配点。例如SlackIntegrationHandler需要SlackService所以AddEventIntegrationServices里有AddSlackService的调用webhook 的具名 HttpClient 定义同理。在AddEventIntegrationServices中创建 handler 的单例services.TryAddSingletonIIntegrationHandlerExampleIntegrationConfigurationDetails, ExampleIntegrationHandler();创建 listener 配置var exampleConfiguration new ExampleListenerConfiguration(globalSettings);把集成加入 RabbitMQ 与 ASB 各自的声明services.AddRabbitMqIntegrationExampleIntegrationConfigurationDetails, ExampleListenerConfiguration(exampleConfiguration);以及services.AddAzureServiceBusIntegrationExampleIntegrationConfigurationDetails, ExampleListenerConfiguration(exampleConfiguration);以上三步与当前仓库中 AddEventIntegrationServices 的实际写法注册 handler 单例 → 构造各 ListenerConfiguration → 按 ASB/RabbitMQ 分别调用Add*Integration泛型扩展完全对应。十一、部署新集成Deploying a New Integration11.1 RabbitMQRabbitMQ 会在队列和交换器首次被代码访问时动态创建它们。因此部署新集成时无需手工创建队列。当然也可以提前创建和配置但不是必须。注意一旦创建完成若需要修改任何配置队列或交换器必须删除后重建。11.2 Azure Service Bus与 RabbitMQ 相反ASB 资源必须在代码访问之前分配不会按需创建。这意味着新集成所需的任何订阅都必须在部署该代码之前在 ASB 中创建好。上文在 Global Settings 与servicebusemulator_config.json中定义的两个订阅需要在部署代码前通过 Azure 门户或 CLI 为对应环境创建ExampleEventSubscriptionName这是从主事件 topic 扇出fan-out的订阅因此它一经声明就会开始接收所有事件这可能在“集成专属 handler 声明并部署”之前形成积压backlog一种规避策略是以假过滤器例如1 0创建该订阅订阅被创建但过滤器确保没有任何消息真正落入该订阅可以部署引用该订阅的代码因为订阅合法存在只是为空代码就位、准备让新集成开始接收消息时移除过滤器即可让订阅恢复接收所有 fan-out 消息。ExampleIntegrationSubscriptionName该订阅必须在新集成代码部署前创建但它不是 fan-out而是基于IntegrationType.ToRoutingKey的过滤器因此在组织拥有激活的配置之前它不会开始接收消息。这意味着提前声明它不会造成积压风险。十二、小结关键路径速查关注点关键类型/文件事件发布入口IEventWriteService / EventIntegrationEventWriteService.cs平台选择与装配EventIntegrationsServiceCollectionExtensions.cs事件层处理器EventIntegrationHandler.cs、EventRepositoryHandler.cs集成层重试RabbitMqIntegrationListenerService.cs、AzureServiceBusIntegrationListenerService.cs消息与退避IntegrationMessage.cs重试/延迟参数GlobalSettings.csEventLogging.MaxRetries、EventLogging.RabbitMq.UseDelayPlugin集成配置管理命令CreateOrganizationIntegrationCommand.cs、UpdateOrganizationIntegrationCommand.cs、DeleteOrganizationIntegrationCommand.cs本地 ASB 资源dev/servicebusemulator_config.json需要说明的适用前提本架构中的 Azure Service Bus 路径对应云端部署RabbitMQ 路径主要面向自托管实例两者的资源创建语义按需创建 vs 预先创建是部署新集成时最需要注意的差异。所有队列/订阅默认值均可通过 Global Settings 或 secrets 覆盖但修改已创建的 RabbitMQ 队列/交换器配置需要先删除再重建。【免费下载链接】serverBitwarden infrastructure/backend (API, database, Docker, etc).项目地址: https://gitcode.com/GitHub_Trending/ser/server创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考