c# - 如何使异步事件连续运行？

Question

我正在创建一个小型控制台应用程序，目前还处于概念验证阶段。这个想法是它将查询 3rd 方 API 以获取他们端的事件。 (那部分一切正常)。

当事件到达时，我们需要将该事件的 json 字符串粘贴到我们这边的 sqlite 数据库中。同时，我需要一个在同一个应用程序中运行的进程，该进程不断在 sqlite 表中查找记录，并在找到记录时进行处理。

这是我目前所拥有的...(同样，这只是一个概念证明，而我正在学习如何设置它的基础知识)...

static void Main(string[] args)
{
    var t = CheckEventsAsync();
    while (true)
    {
        System.Threading.Thread.Sleep(5000);
        DataAccess.InsertData("INSERT INTO events(value) VALUES ('this is an event');");
    }
}

static private async Task CheckEventsAsync()
{
    DataTable dt = await DataAccess.LoadDataAsync("SELECT * FROM events");
    foreach (var dr in dt.Rows)
    {
        Console.WriteLine("got an event!");
    }
}

问题是 CheckEventsAsync() 只触发一次，然后当 5 秒计时器关闭时，一条记录被添加到表中，没有任何反应。

我是否需要使用 EventHandler(带有委托(delegate)等)来完成这项工作？或者是否有可能不这样做？

谢谢!

哦，差点忘了...(这样做的原因是如果有数据它会返回一个数据表，如果没有数据则返回 null。但我不确定我这样做是否符合总体目标)...

    public static async Task<DataTable> LoadDataAsync(string txtQuery)
    {
        var dataTable = new DataTable();
        var connection = new SQLiteConnection(_source);
        using (connection)
        {
            connection.Open();
            var command = connection.CreateCommand();
            command.CommandText = txtQuery;
            var task = await command.ExecuteReaderAsync();
            dataTable = new DataTable();
            dataTable.Load(task);
            connection.Close();
        }
        if (dataTable != null & dataTable.Rows.Count > 0)
            return dataTable;
        else
            return null;
    }

编辑

从目前的答案来看，我提出的问题做得很糟糕。我需要运行两个进程:

1. 第一个进程将监听第 3 方事件并将它们添加到 sqlite 表中。这是用 5 秒计时器模拟的。但实际上，While(true) 循环中不会有计时器。它将是一个 WebSocket 监听器。
2. 第二个进程将监视数据库中的记录并处理它们。因此，CheckEventsAsync() 需要是一个无需调用方法即可触发的监听器。

两者需要在同一个应用程序中运行，但彼此独立。

我希望我说得有道理？

Answer 1

为此使用 Microsoft 的 Reactive Framework:

static readonly BlockingCollection<ItemToProcess> _queue = new BlockingCollection<ItemToProcess>(new ConcurrentQueue<ItemToProcess>());

static void Main(string[] args)
{
    IDisposable inserts =
        Observable
            .Interval(TimeSpan.FromSeconds(5.0))
            .Subscribe(_ =>
            {
                _queue.Add(new ItemToProcess() { Value = "value" });
            });

    IDisposable checks =
        new CompositeDisposable(
            Disposable.Create(() => _queue.CompleteAdding()),
            _queue
                .GetConsumingEnumerable()
                .ToObservable(Scheduler.Default)
                .Subscribe(item =>
                {
                    Console.WriteLine(item.Value);
                }));
}

这会处理所有线程并使关闭应用程序变得容易。只需在退出前调用 inserts.Dispose() 和 checks.Dispose()。

NuGet“System.Reactive”获取位。