如何推迟事件触发/接收

Question

SUM：如何在 azure 事件网格中推迟事件触发或事件接收？

我设计的系统需要对低频 OBJECT 状态（创建、启动、检查长时间处于启动状态、结束）做出反应。它看起来像是事件处理的候选者。我想用天蓝色的功能来实现它...... 问题是我需要对处于 STARTED 状态的 10 分钟（可配置）的 OBJECT 做出反应。在事件发生后 10 分钟做出反应。如何延迟事件触发？如何安排事件触发？如何等待事件？ 我正在寻找不会消耗我的付费资源（函数处理时间）的解决方案。 知道如何解决这个问题吗？谢谢。

功能1：

• 时间触发（轮询）
• 检查条件
  • 可以创建 OBJECT-created 事件

功能2：

• http 触发
• 用户从 UI 启动 OBJECT
  • 创建 OBJECT-started 事件但在 10 分钟后触发它？ （在函数代码中休眠/定时器会消耗10分钟的资源并以过高的成本结束）

功能3：？？？

• ???
• 检查状态并检测 10 分钟延迟
  • 触发 OBJECT-10 分钟开始的事件

...

Answer 1

在事件发生 10 分钟后做出反应。

首先我要强调的是，事件应该代表过去发生的事情，而不是将来发生的事情。事件的期望是通知发生了什么。

其次，如果您对需要发生的事情有预期，那么命令比事件要好。在这种情况下，您希望工作在 10 分钟内发生在 Blob 上，该 Blob 将处于实际可用于工作的状态。我将通过向队列（Azure 服务总线）发送延迟消息来响应该事件，其中包含在 blob 上工作所需的信息。这样，您就可以通过准备未来/延迟的工作项来响应事件。

Answer 2

如何在 azure 事件网格中推迟事件触发或事件接收？

Azure 事件网格 (AEG) 没有内置此功能，但是很容易使用 Azure 服务总线 (ASB) 实体来扩展它以处理延迟（计划）消息，就像 Sean 在回答中提到的那样。

以下屏幕 sn-p 显示了 Push-and-Pull with delay 订阅者的概念：

事件消息被推送到 ASB 主题中，并根据其订阅规则，将事件消息作为预定消息转发到队列实体。

主题订阅需要设置以下属性：

• 转发到

  name of the queue/topic entity
  

• $默认规则

  过滤器：

  1=1 
  

  动作（例如 10 分钟）：

  SET sys.TimeToLive = '00:10:00';
  SET EnqueuedTimeUtc = sys.EnqueuedTimeUtc;
  SET ScheduledEnqueueTimeUtc = sys.ExpiresAtUtc;
  SET sys.ScheduledEnqueueTimeUtc = sys.ExpiresAtUtc;
  SET sys.Label = 'Delay';
  SET sys.TimeToLive = '01:00:00';
  

目标队列：

 EnableDeadLetteringOnMessageExpiration = true

根据上面的设置，队列中的调度消息必须在'01:00:00'等TTL内消费，否则将消息发送到DLQ。肖恩评论中的更多细节。

使用ServiceBusTrigger函数，可以像AEG订阅者一样，以透明的方式从队列中拉出延迟的事件消息。

在这种情况下，当延迟事件被发送回 AEG 以进行 Fan-Out 分发并使用 Push-and-Push 模式时，以下示例显示ServiceBusTrigger 的这个实现，输出绑定到 AEG 自定义主题：

运行.csx：

#r "Newtonsoft.Json"
#r "Microsoft.Azure.EventGrid"
#r "Microsoft.Azure.ServiceBus"


using System.Net;
using System.Text;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Primitives;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Microsoft.Azure.EventGrid.Models;
using Microsoft.Azure.ServiceBus;


public static async Task Run(Message queueItem, IAsyncCollector<EventGridEvent> outputEvents, ILogger log)
{   
    string jsontext = JToken.Parse(Encoding.UTF8.GetString(queueItem.Body)).ToString(Formatting.Indented); 
    log.LogInformation(jsontext);

    EventGridEvent eventGridEvent = JsonConvert.DeserializeObject<EventGridEvent>(jsontext);
    eventGridEvent.Topic = null;
    eventGridEvent.Subject += "/delayed";
    await outputEvents.AddAsync(eventGridEvent);

    await Task.CompletedTask;
}

function.json:

{
  "bindings": [
    {
      "name": "queueItem",
      "type": "serviceBusTrigger",
      "direction": "in",
      "queueName": "aeg",
      "connection": "rk2016_SERVICEBUS"
    },
    {
      "type": "eventGrid",
      "direction": "out",
      "name": "outputEvents",
      "topicEndpointUri": "AEG_TOPIC_XX_ENDPOINT",
      "topicKeySetting": "AEG_TOPIC_XX_KEY"
    }
  ]
}

如您所见，主题属性已修改为后缀/delayed，用于过滤目的，例如避免循环等。