asynchronous 相关问题

考虑我有大量整数，并且此方法模拟异步外部 api 数据检索，它可以为某些特定的未知输入返回空响应：公共静态单声道考虑到我有大量的整数，并且此方法模拟异步外部 api 数据检索，它可以为某些特定的未知输入返回空响应： public static Mono<String> getApiData(int i) { if (i == 3) return Mono.empty(); // i'm using 3 just as an example return Mono.just(String.valueOf(i * 2)); } 这些方法将根据结果执行 getApiData 输出： // when getApiData returns non empty mono public static Mono<Boolean> updateDatabaseWithApiData(int apiInput, String apiOutput) { System.out.println(apiInput + " -> " + apiOutput); // lots of unrelated logic return Mono.just(true); } // when getApiData returns empty mono public static Mono<Boolean> logFailure(int apiInput) { System.out.println(apiInput + " -> failure"); // registering errors logs return Mono.just(false); } 用这个我想编写一个像这样的方法Mono<Boolean> processFluxUntilFailure(Flux<Integer> flux)，它对每个元素应用getApiData，并且在发生故障时停止。那么如果至少有一个元素达到updateDatabaseWithApiData，则返回Mono.just(true)，否则返回Mono.just(false)。所以我会得到这个输出： public static void main(String[] args) { Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5); processFluxUntilFailure(flux).subscribe(value -> System.out.println("result " + value)); } 所需输出： 1 -> 2 2 -> 4 3 -> failure result true 因为我们已经处理了（至少 1）2 个成功的元素。考虑到：这是我真正问题的简化版本我无法预测数据何时会为空 getApiData 我无法改变所描述的方法，只能processFluxUntilFailure 我试过这个： public static Mono<Boolean> processFluxUntilFailure(Flux<Integer> flux) { return flux.flatMap(apiInput -> getApiData(apiInput) .flatMap(apiOutput -> updateDatabaseWithApiData(apiInput, apiOutput)) .switchIfEmpty(Mono.defer(() -> logFailure(apiInput))) ) .reduce((b1, b2) -> b1 || b2); } 这导致了 1 -> 2 2 -> 4 3 -> failure 4 -> 8 5 -> 10 result true 我怎样才能从这次尝试中获得我想要的输出？换句话说，如果满足某些异步条件，我如何“停止”flatMap？如果有任何不那么冗长的方法，我很乐意： public static Mono<Boolean> processAndUpdate(Flux<Integer> flux) { return flux .flatMap(apiInput -> getApiData(apiInput) .flatMap(apiOutput -> updateDatabaseWithApiData(apiInput, apiOutput)) .switchIfEmpty(Mono.defer(() -> logFailure(apiInput))) ).<Boolean>handle((b, sink) -> { if (b) sink.next(true); else sink.complete(); }) .defaultIfEmpty(false) .reduce((b1, b2) -> b1 || b2); } 给我这个输出： 1 -> 2 2 -> 4 3 -> failure result true 显然 flatMap 和 handle 并没有急于评估，将它们放在一起就成功了

java asynchronous project-reactor flatmap

回答 1 投票 0

Rust 在不使用 Arc/Mutex 的情况下从多个线程访问对象

我正在尝试用 Rust 编写一个下载器。我面临的问题是我正在尝试实现一个暂停功能，以便下载方法可以继续进行，并且可以调用另一个暂停方法...

asynchronous rust concurrency

回答 1 投票 0

如何在单独线程内的同步函数中向用户发送消息？

我目前正在开发一个电报机器人。我想创建一个完全像 CLI 游戏一样的游戏，但您需要向 telegram 机器人发送消息才能玩它。我遇到了一个问题，我的游戏......

python asynchronous python-telegram-bot

回答 1 投票 0

Asyncpraw 基本 reddit API 示例在执行后立即抛出错误

我刚刚开始尝试使用 AsyncPraw Reddit API 包装器创建一个机器人。我复制粘贴了文档中的代码示例，但我无法在没有警告的情况下运行它。通过尝试...

python asynchronous async-await praw asyncpraw

回答 1 投票 0

有什么方法可以使通知去抖观察程序异步吗？

我在这里的通知仓库中问了同样的问题有没有办法让 debounce-watcher 异步？但是使用 tokio::sync::mpsc::channel 时，程序仍然停留在 while let Some(res) = rx.recv().a...

asynchronous rust rust-tokio notify

回答 2 投票 0

ToArrayAsync() 抛出“源 IQueryable 未实现 IAsyncEnumerable”

我在 ASP.NET Core 上有一个 MVC 项目，我的问题与 IQueryable 和异步有关。我在 IQueryable 中编写了以下搜索方法：私有 IQueryable 我在 ASP.NET Core 上有一个 MVC 项目，我的问题与 IQueryable 和异步有关。我在IQueryable<T>中写了以下搜索方法： private IQueryable<InternalOrderInfo> WhereSearchTokens(IQueryable<InternalOrderInfo> query, SearchToken[] searchTokens) { if (searchTokens.Length == 0) { return query; } var results = new List<InternalOrderInfo>(); foreach (var searchToken in searchTokens) { //search logic, intermediate results are being added to `results` using `AddRange()` } return results.Count != 0 ? results.Distinct().AsQueryable() : query; } 我在方法中称之为ExecuteAsync()： public async Task<GetAllInternalOrderInfoResponse> ExecuteAsync(GetAllInternalOrderInfoRequest request) { //rest of the code if (searchTokens != null && searchTokens.Any()) { allInternalOrderInfo = WhereSearchTokens(allInternalOrderInfo, searchTokens); } var orders = await allInternalOrderInfo.Skip(offset).Take(limit).ToArrayAsync(); //rest of the code } 当我测试这个时，我在调用 ToArrayAsync() 的地方收到 InvalidOperationException 源 IQueryable 没有实现 IAsyncEnumerable。只有实现 IAsyncEnumerable 的源才能用于实体框架异步操作。我已将 ToArrayAsync() 更改为 ToListAsync() 但没有任何改变。我已经搜索这个问题一段时间了，但已解决的问题主要与 DbContext 和实体创建有关。该项目没有安装 EntityFramework，由于应用程序架构的原因，最好不要安装。希望有人知道在我的情况下该怎么做。我发现我必须做更多的工作才能让事情顺利进行： namespace TestDoubles { using Microsoft.EntityFrameworkCore.Query.Internal; using System; using System.Collections.Generic; using System.Linq; using System.Linq.Expressions; using System.Threading; using System.Threading.Tasks; public static class AsyncQueryable { /// <summary> /// Returns the input typed as IQueryable that can be queried asynchronously /// </summary> /// <typeparam name="TEntity">The item type</typeparam> /// <param name="source">The input</param> public static IQueryable<TEntity> AsAsyncQueryable<TEntity>(this IEnumerable<TEntity> source) => new AsyncQueryable<TEntity>(source ?? throw new ArgumentNullException(nameof(source))); } public class AsyncQueryable<TEntity> : EnumerableQuery<TEntity>, IAsyncEnumerable<TEntity>, IQueryable<TEntity> { public AsyncQueryable(IEnumerable<TEntity> enumerable) : base(enumerable) { } public AsyncQueryable(Expression expression) : base(expression) { } public IAsyncEnumerator<TEntity> GetEnumerator() => new AsyncEnumerator(this.AsEnumerable().GetEnumerator()); public IAsyncEnumerator<TEntity> GetAsyncEnumerator(CancellationToken cancellationToken = default) => new AsyncEnumerator(this.AsEnumerable().GetEnumerator()); IQueryProvider IQueryable.Provider => new AsyncQueryProvider(this); class AsyncEnumerator : IAsyncEnumerator<TEntity> { private readonly IEnumerator<TEntity> inner; public AsyncEnumerator(IEnumerator<TEntity> inner) => this.inner = inner; public void Dispose() => inner.Dispose(); public TEntity Current => inner.Current; public ValueTask<bool> MoveNextAsync() => new ValueTask<bool>(inner.MoveNext()); #pragma warning disable CS1998 // Nothing to await public async ValueTask DisposeAsync() => inner.Dispose(); #pragma warning restore CS1998 } class AsyncQueryProvider : IAsyncQueryProvider { private readonly IQueryProvider inner; internal AsyncQueryProvider(IQueryProvider inner) => this.inner = inner; public IQueryable CreateQuery(Expression expression) => new AsyncQueryable<TEntity>(expression); public IQueryable<TElement> CreateQuery<TElement>(Expression expression) => new AsyncQueryable<TElement>(expression); public object Execute(Expression expression) => inner.Execute(expression); public TResult Execute<TResult>(Expression expression) => inner.Execute<TResult>(expression); public IAsyncEnumerable<TResult> ExecuteAsync<TResult>(Expression expression) => new AsyncQueryable<TResult>(expression); TResult IAsyncQueryProvider.ExecuteAsync<TResult>(Expression expression, CancellationToken cancellationToken) => Execute<TResult>(expression); } } } 这使我能够编写这样的测试： [TestCase("", 3, 5)] [TestCase("100", 2, 4)] public async Task GetOrderStatusCounts_ReturnsCorrectNumberOfRecords(string query, int expectedCount, int expectedStatusProductionCount) { // omitted CreateOrder helper function const int productionStatus = 6; const int firstOtherStatus = 5; const int otherOtherStatus = 7; var items = new[] { CreateOrder(1, "100000", firstOtherStatus, 1), CreateOrder(2, "100000", firstOtherStatus, 4), CreateOrder(3, "100000", productionStatus, 4), CreateOrder(4, "100001", productionStatus, 4), CreateOrder(5, "100100", productionStatus, 4), CreateOrder(6, "200000", otherOtherStatus, 4), CreateOrder(7, "200001", productionStatus, 4), CreateOrder(8, "200100", productionStatus, 4) }.AsAsyncQueryable(); // this is where the magic happens var mocker = new AutoMocker(); // IRepository implementation is also generic and calls DBCntext // for easier testing mocker.GetMock<IRepository<Order>>() .Setup(m => m.BaseQuery() .Returns(items); // the base query is extended in the system under test. // that's the behavior I'm testing here var sut = mocker.CreateInstance<OrderService>(); var counts = await sut.GetOrderStatusCountsAsync(4, query); counts.Should().HaveCount(expectedCount); counts[OrderStatus.Production].Should().Be(expectedStatusProductionCount); } 我编写了一个 ICollection 扩展 AsAsyncQueryable，我在测试中使用它 using System; using System.Collections; using System.Collections.Generic; using System.Linq; using System.Linq.Expressions; using System.Threading; using System.Threading.Tasks; namespace Whatevaaaaaaaa { public static class ICollectionExtensions { public static IQueryable<T> AsAsyncQueryable<T>(this ICollection<T> source) => new AsyncQueryable<T>(source.AsQueryable()); } internal class AsyncQueryable<T> : IAsyncEnumerable<T>, IQueryable<T> { private IQueryable<T> Source; public AsyncQueryable(IQueryable<T> source) { Source = source; } public Type ElementType => typeof(T); public Expression Expression => Source.Expression; public IQueryProvider Provider => new AsyncQueryProvider<T>(Source.Provider); public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new AsyncEnumeratorWrapper<T>(Source.GetEnumerator()); } public IEnumerator<T> GetEnumerator() => Source.GetEnumerator(); IEnumerator IEnumerable.GetEnumerator() => GetEnumerator(); } internal class AsyncQueryProvider<T> : IQueryProvider { private readonly IQueryProvider Source; public AsyncQueryProvider(IQueryProvider source) { Source = source; } public IQueryable CreateQuery(Expression expression) => Source.CreateQuery(expression); public IQueryable<TElement> CreateQuery<TElement>(Expression expression) => new AsyncQueryable<TElement>(Source.CreateQuery<TElement>(expression)); public object Execute(Expression expression) => Execute<T>(expression); public TResult Execute<TResult>(Expression expression) => Source.Execute<TResult>(expression); } internal class AsyncEnumeratorWrapper<T> : IAsyncEnumerator<T> { private readonly IEnumerator<T> Source; public AsyncEnumeratorWrapper(IEnumerator<T> source) { Source = source; } public T Current => Source.Current; public ValueTask DisposeAsync() { return new ValueTask(Task.CompletedTask); } public ValueTask<bool> MoveNextAsync() { return new ValueTask<bool>(Source.MoveNext()); } } } 如果您不打算更改设计 - 您有多种选择： 1) 将 AsQueryable 更改为另一个返回 IQueryable 的方法，该方法也实现了 IDbAsyncEnumerable。例如，您可以扩展 EnumerableQuery （由 AsQueryable 返回）： public class AsyncEnumerableQuery<T> : EnumerableQuery<T>, IDbAsyncEnumerable<T> { public AsyncEnumerableQuery(IEnumerable<T> enumerable) : base(enumerable) { } public AsyncEnumerableQuery(Expression expression) : base(expression) { } public IDbAsyncEnumerator<T> GetAsyncEnumerator() { return new InMemoryDbAsyncEnumerator<T>(((IEnumerable<T>) this).GetEnumerator()); } IDbAsyncEnumerator IDbAsyncEnumerable.GetAsyncEnumerator() { return GetAsyncEnumerator(); } private class InMemoryDbAsyncEnumerator<T> : IDbAsyncEnumerator<T> { private readonly IEnumerator<T> _enumerator; public InMemoryDbAsyncEnumerator(IEnumerator<T> enumerator) { _enumerator = enumerator; } public void Dispose() { } public Task<bool> MoveNextAsync(CancellationToken cancellationToken) { return Task.FromResult(_enumerator.MoveNext()); } public T Current => _enumerator.Current; object IDbAsyncEnumerator.Current => Current; } } 然后你改变 results.Distinct().AsQueryable() 到 new AsyncEnumerableQuery<InternalOrderInfo>(results.Distinct()) 之后，ToArrayAsync将不再抛出异常（显然你可以像AsQueryable一样创建自己的扩展方法）。 2）更改ToArrayAsync部分： public static class EfExtensions { public static Task<TSource[]> ToArrayAsyncSafe<TSource>(this IQueryable<TSource> source) { if (source == null) throw new ArgumentNullException(nameof(source)); if (!(source is IDbAsyncEnumerable<TSource>)) return Task.FromResult(source.ToArray()); return source.ToArrayAsync(); } } 并使用 ToArrayAsyncSafe 代替 ToArrayAsync，如果 IQueryable 不是 IDbAsyncEnumerable，它将回退到同步枚举。在您的情况下，只有当查询实际上是内存中列表而不是查询时才会发生这种情况，因此异步执行无论如何都是没有意义的。对于 EF Core： public static class QueryableExtensions { public static IQueryable<T> AsAsyncQueryable<T>(this IEnumerable<T> input) { return new NotInDbSet<T>( input ); } } public class NotInDbSet< T > : IQueryable<T>, IAsyncEnumerable< T >, IEnumerable< T >, IEnumerable { private readonly List< T > _innerCollection; public NotInDbSet( IEnumerable< T > innerCollection ) { _innerCollection = innerCollection.ToList(); } public IAsyncEnumerator< T > GetAsyncEnumerator( CancellationToken cancellationToken = new CancellationToken() ) { return new AsyncEnumerator( GetEnumerator() ); } public IEnumerator< T > GetEnumerator() { return _innerCollection.GetEnumerator(); } IEnumerator IEnumerable.GetEnumerator() { return GetEnumerator(); } public class AsyncEnumerator : IAsyncEnumerator< T > { private readonly IEnumerator< T > _enumerator; public AsyncEnumerator( IEnumerator< T > enumerator ) { _enumerator = enumerator; } public ValueTask DisposeAsync() { return new ValueTask(); } public ValueTask< bool > MoveNextAsync() { return new ValueTask< bool >( _enumerator.MoveNext() ); } public T Current => _enumerator.Current; } public Type ElementType => typeof( T ); public Expression Expression => Expression.Empty(); public IQueryProvider Provider => new EnumerableQuery<T>( Expression ); } 对于 EFCore 有点晚了，但对于其他希望解决此类问题的人来说，可能的解决方案之一是更改代码以这种方式使用 Task.FromResult() 方法： var result= await allInternalOrderInfo.Skip(offset).Take(limit); var orders = await Task.FromResult(result.ToArray()); AsQueryable()不会将result列表转换为实体框架IQueryable。正如错误所述，与 IQueryable 一起使用的 ToArrayAsync() 应该实现 IAsyncEnumerable，这不是 AsQueryable 将返回的内容。您可以在此处阅读有关 AsQueryable 在可枚举上的使用的更多信息。正如 @Titian Cernicova-Dragomir 所指出的，异常意味着 List<InternalOrderInfo> 没有实现 IAsyncEnumerable 但这里有一个逻辑/设计错误。如果您的方法适用于 IQueryable 并返回 IQueryable，则它应该像 IQueryable 一样使用它，而不是像假设集合位于应用程序内存中的 IEnumarable 那样。您确实需要详细了解 IQueryable 和 IEnumarable 之间的区别以及应该从该方法返回什么。一个好的开始点是阅读答案这里和这里因此，由于您已经在 WhereSearchTokens 方法中甚至之前从数据库中获取了结果，因此没有理由对数据库进行异步请求，这将由 ToArrayAsync 完成并返回 IQueryable。您有两个选择： 1）如果您的InternalOrderInfo集合在WhereSearchTokens之前从数据库获取到内存中，则使您的所有操作处于同步模式，即调用ToArray而不是ToArrayAsync，并从两者返回IEnumerable而不是Taks<IQueryable> WhereSearchTokens 和 ExecuteAsync。 2）如果您的InternalOrderInfo集合是在WhereSearchTokens内获取的，并且您想要对数据库执行异步请求，则只需在//search logic, intermediate results are being added to results using AddRange()中的某个位置调用异步EF API，然后再次返回Taks<IEnumerable>而不是Taks<IQueryable>来自WhereSearchTokens 错误消息： System.InvalidOperationException：源“IQueryable”未实现“IAsyncEnumerable”。只有实现“IAsyncEnumerable”的源才能用于实体框架异步操作。对于我的情况，解决方案：当您模拟 dbContext 并将数据从mockSet传递到上下文时，将 .Returns 更改为 .ReturnsDbSet 示例： var mockContext = new Mock<IWebApiDbContext>(); mockContext.Setup(m => m.User).ReturnsDbSet(mockSet.Object); 完整代码模拟数据库： var mockSet = new Mock<DbSet<User>>(); mockSet.As<IDbAsyncEnumerable<User>>() .Setup(m => m.GetAsyncEnumerator()) .Returns(new TestDbAsyncEnumerator<User>(data.GetEnumerator())); mockSet.As<IQueryable<User>>() .Setup(m => m.Provider) .Returns(new TestDbAsyncQueryProvider<User>(data.Provider)); mockSet.As<IQueryable<User>>().Setup(m => m.Expression).Returns(data.Expression); mockSet.As<IQueryable<User>>().Setup(m => m.ElementType).Returns(data.ElementType); mockSet.As<IQueryable<User>>().Setup(m => m.GetEnumerator()).Returns(() => data.GetEnumerator()); var mockContext = new Mock<IWebApiDbContext>(); mockContext.Setup(m => m.User).ReturnsDbSet(mockSet.Object); 最好使用 IAsyncEnumerable<T> 和 IQueryable<T> 来实现集合，而不是创建自己的 ToListAsync 扩展。您无法在库中应用扩展。对于 EF Core 5 及更高版本，请检查此实现和测试。简短版： public sealed class FixedQuery<T> : IAsyncEnumerable<T>, IQueryable<T> { public static readonly IQueryable<T> Empty = Create(ArraySegment<T>.Empty); public static IQueryable<T> Create(params T[] items) { return Create((IEnumerable<T>)items); } public static IQueryable<T> Create(IEnumerable<T> items) { return new FixedQuery<T>(items ?? ArraySegment<T>.Empty).AsQueryable(); } private readonly IQueryable<T> _items; private FixedQuery(IEnumerable<T> items) { _items = (items ?? throw new ArgumentNullException(nameof(items))).AsQueryable(); } #pragma warning disable CS1998 public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default) #pragma warning restore CS1998 { foreach (var item in _items) { yield return item; } } public IEnumerator<T> GetEnumerator() { return _items.GetEnumerator(); } IEnumerator IEnumerable.GetEnumerator() { return GetEnumerator(); } public Type ElementType => _items.ElementType; public Expression Expression => _items.Expression; public IQueryProvider Provider => _items.Provider; } 我有同样的错误消息。我知道你的问题表明你不想安装实体框架，但就我而言，其他读者提出这个问题并且没有类似的约束，改变 using System.Data.Entity; 到 using Microsoft.EntityFrameworkCore; 为我工作。使用此扩展MockQueryable //1 - create a List<T> with test items var users = new List<UserEntity>() { new UserEntity{LastName = "ExistLastName", DateOfBirth = DateTime.Parse("01/20/2012")}, ... }; //2 - build mock by extension var mock = users.BuildMock(); //3 - setup the mock as Queryable for Moq _userRepository.Setup(x => x.GetQueryable()).Returns(mock); //3 - setup the mock as Queryable for NSubstitute _userRepository.GetQueryable().Returns(mock); //3 - setup the mock as Queryable for FakeItEasy A.CallTo(() => userRepository.GetQueryable()).Returns(mock);

c# entity-framework asynchronous asp.net-core asp.net-core-mvc

回答 11 投票 0

即使实际实现工作正常，RestSharp IRestClient 的模拟也会失败并出现 NullReferenceException

我已经为此绞尽脑汁好几天了，需要向社区询问这个问题。我已经用 C# 开发了一段时间，但对于实际使用异步还很陌生。我正在制作一个 Rest API 客户端并且

c# .net asynchronous nunit restsharp

回答 1 投票 0

如何在下一个“then”之前等待多个异步调用在“then”中完成？ [重复]

我进行异步调用来获取一些数据。然后，我需要进行 3 个异步调用。这 3 个电话结束后，我想做点别的事情。我试过这个：让outerApi； getApi(url).then(函数(api) {

javascript asynchronous promise

回答 1 投票 0

PowerShell 中作业有新数据时异步触发函数

我希望控制台在作业有新数据（来自以 Start-ThreadJob 启动的作业）时触发函数，即每当 .HasMoreData 为 True 时。我希望 Register-ObjectEvent 能帮助我