在表存储上执行事件溯源投影

Question

我正在创建一个小型的事件源式函数应用程序，其中每次调用函数都会将事件写入表存储。这种事件的一个例子是：

+------------+---------------+-----------------+
|   Event    |   Timestamp   |   Destination   |
+------------+---------------+-----------------+
| Connect    | 7/1/2019 4:52 | sftp.alex.com   |
| Disconnect | 7/1/2019 4:53 | sftp.liza.com   |
| Connect    | 7/1/2019 4:54 | sftp.yomama.com |
| Connect    | 7/1/2019 4:54 | sftp.alex.com   |
| Connect    | 7/1/2019 4:59 | sftp.liza.com   |
| Disconnect | 7/1/2019 4:59 | sftp.alex.com   |
| Disconnect | 7/1/2019 4:59 | sftp.yomama.com |
| Connect    | 7/1/2019 5:03 | sftp.alex.com   |
+------------+---------------+-----------------+

如何在此表上创建投影？

我需要回答的主要问题是：

每个目的地目前有多少个连接？

Answer 1

我想表中会有很多记录，并且遍历所有记录不是一种选择。
所以这里有几个想法：

1. 你不能只跟踪连接数吗？

   那将是最简单的解决方案。我不知道您的应用程序以及它如何与Azure通信，但至少有触发器（尽管根据支持的绑定表判断，您将需要使用一些额外的服务……例如Queue storage）。在它们中，您应该能够在单独的表中存储到每个目的地的当前连接数，随着Connect事件增加而减少Disconnect。

   但是，如果您只有一个编写器（与Azure通信的单个服务器），您可以在代码内部跟踪连接。

   您还可以将当前连接到表的数量保存在一个额外的字段中。作为奖励，您将能够在过去的任何给定时间立即获得大量连接（以内存为代价）。

2. 当您谈论事件溯源时……那么也许您应该再次使用它？想法仍然是相同的：你跟踪Connect和Disconnect事件，但在一些外部接收器。当您正在编写事件溯源风格的函数应用程序时，我相信创建一个应该很容易。而且您不必依赖额外的Azure服务。

   那么与第一个想法的唯一区别是，如果接收器死亡或断开连接或其他事情 - 只需记住它接收到的最后一个事件，当接收器重新上线时，只迭代年轻的事件。

   您应该记住的最后收到的事件（加上计数器）本质上是其他人在评论中谈论的快照。

Answer 2

投影应该与事件流分离，因为它们是业务驱动的，而事件流纯粹是一个技术方面。

我假设您将使用 SQL 来持久化预测以简化答案，但任何键/值数据存储都可以。

您可以创建DestinationEvents具有以下结构的表：

+------------------+-----------------+-------------------+
|   Destination    |   Connections   |   Disconnections  |
+------------------+-----------------+-------------------+
| sftp.alex.com    |        3        |        1          |
| sftp.liza.com    |        1        |        1          |
+------------------+-----------------+-------------------+

通过适当的索引，这应该可以提供快速读取和写入。为了获得额外的速度，请考虑使用 Redis 之类的东西来缓存您的预测。

棘手的一点是在解决方案设计中，您希望它可以扩展。一种幼稚的方法可能是为每次写入事件流设置 SQL 触发器，但如果您有大量写入，这会减慢您的速度。

如果您想要可扩展性，您需要开始考虑预算（时间和金钱）和业务需求。预测是否需要实时可用？

• 如果不是，您可以有一个计划的过程，以特定的时间间隔计算/更新预测：每天、每小时、每周等。
• 如果是，您需要开始研究队列/消息代理（RabbitMQ、Kafka 等）。现在您正在进入生产者/消费者逻辑。您的应用程序是生产者，它发布事件。EventStream 和 Projections 存储是消费者，它们侦听、转换和持久化传入的事件。Queue/MessageBroker 本身可以替换您的 Event Stream 表，使用 Kafka 很容易。

如果您只是想学习，请首先使用Dictionary<string, (int Connections, int Disconnections)>Destination 作为 Key 并且(int Connections, int Disconnections)是元组/类来定义内存中的投影存储。

如果你想支持其他投影，内存中的存储可以是一个Dictionary<string, Dictionary<string, (int Connections, int Disconnections)>>，其中外部字典 Key 是投影名称。

Answer 3

基本思想是在聚合上重播事件以获得当前状态。下面是说明它的代码。警告：这不是生产代码，甚至无法编译。

public class ConnectionCounters
{
    private Dictionary<string, ConnectionCounter> _counters = new Dictionary<string, ConnectionCounter>();

    public IEnumerable<ConnectionCounter> GetCounters()
    {
        return _counters.Values;
    }

    public void Handle(ConnectionEvent @event)
    {
        var counter = GetOrCreateCounter(@event.Destination);
        if (@event is ConnectEvent)
            counter.ConnectionCount += 1;
        if (@event is DisconnectEvent)
            counter.ConnectionCount -= 1;
    }

    private ConnectionCounter GetOrCreateCounter(string destination)
    {
        if (_counters.ContainsKey(destination))
            return _counters[destination];

        var counter = new ConnectionCounter() { Destination = destination };
        _counters[destination] = counter;
        return counter;
    }
}

public class ConnectionCounter
{
    public string Destination { get; set; }
    public int ConnectionCount { get; set; }
}

public class ConnectEvent : ConnectionEvent { }

public class DisconnectEvent : ConnectionEvent { }

public class ConnectionEvent 
{
    public string Destination { get; set; }
}

// .....

private ConnectionCounters _connectionCounters = new ConnectionCounters();
public void Main()
{
    var events = ReadEvents(); // read events somehow
    foreach (var @event in events)
    {
        _connectionCounters.Handle(@event);
    }

    foreach (var counter in _connectionCounters.GetCounters())
        Console.WriteLine($"{counter.Destination} has {counter.ConnectionCount} connections.")
}