c# - 使用 Reactive Extensions - 这是异步的吗？

Question

尝试将现有的数据访问代码转换为异步并遇到 Rx，因为您无法返回 Task<IEnumerable<T>>用yield return在你的方法体中。

我写了这篇文章，但不确定它是异步的，所以非常感谢您的指点

public class EmployeeRepository : IEmployeeRepository
{
    public IAsyncEnumerable<Employee> GetEmployees()
    {
        return Enumerable().ToAsyncEnumerable();
    }

    private IEnumerable<Employee> Enumerable()
    {
        using (var connection = new SqlConnection(ConfigurationManager.ConnectionStrings["DBConnString"].ConnectionString))
        {
            connection.Open();
            using (var command = new SqlCommand(@"SELECT * FROM EMPLOYEES", connection))
            {
                using (var reader = command.ExecuteReader())
                {
                    while (reader.Read())
                    {
                        yield return
                            new Employee()
                                {
                                    Id = ReadField<int>(reader, "Id"),
                                    Name = ReadField<string>(reader, "Name")
                                };
                    }
                }
            }
        }
    }

    private static T ReadField<T>(IDataRecord reader, string fieldName)
    {
        var value = reader[fieldName];
        return value == DBNull.Value ? default(T) : (T)value;
    }
}

Answer 1

这不是异步的。 ToAsyncEnumerable 创建一个简单的适配器，它会在每次调用 MoveNext 时阻塞。返回这样一个异步适配器是不好的做法，就像执行 Task.Run(() => BlockingMethod()) 一样。它向用户隐藏了实现效率低下的情况，如果他们知道存在这种情况，他们可能能够以更好的方式解决。

IAsyncEnumerable 没有语言集成的 yield 特性，但可以模拟。我 have code to do it ，但公平警告这会产生一些开销:

IAsyncEnumerable<Employee> async = AsyncEnumerableEx.Create<Employee>(
                                                  async (y, cancellationToken) =>
{
    using (var connection = new SqlConnection(ConfigurationManager
                            .ConnectionStrings["DBConnString"].ConnectionString))
    {
        await connection.OpenAsync(cancellationToken);
        using (var command = new SqlCommand(@"SELECT * FROM EMPLOYEES",
                                            connection))
        {
            using (var reader = await
                                   command.ExecuteReaderAsync(cancellationToken))
            {
                while (await reader.ReadAsync(cancellationToken))
                {
                    await y.YieldReturn(new Employee()
                    {
                        Id = ReadField<int>(reader, "Id"),
                        Name = ReadField<string>(reader, "Name")
                    });
                }
            }
        }
    }
});

如果你想使用实际的 Rx，它内置了一个几乎相同的 Observable.Create 实用程序。由于减少了一些等待开销，它会稍微更有效率。

IObservable<Employee> async = Observable.Create<Employee>(
                                                    async (obs, cancellationToken) =>
{
    using (var connection = new SqlConnection(ConfigurationManager
                            .ConnectionStrings["DBConnString"].ConnectionString))
    {
        await connection.OpenAsync(cancellationToken);
        using (var command = new SqlCommand(@"SELECT * FROM EMPLOYEES",
                                            connection))
        {
            using (var reader = await
                                   command.ExecuteReaderAsync(cancellationToken))
            {
                while (await reader.ReadAsync(cancellationToken))
                {
                    obs.OnNext(new Employee()
                    {
                        Id = ReadField<int>(reader, "Id"),
                        Name = ReadField<string>(reader, "Name")
                    });
                }
            }
        }
    }
});