代码之家 › 专栏 › 技术社区 › ZorgoZ

代码从捕获中丢失?

.net-4.6.2 system.reactive async-await sql-server c#

ZorgoZ · 技术社区 · 6 年前

我有一个后台任务,每200毫秒轮询一个SQL Server数据库。

代码如下所示:

listener = await Task.Factory.StartNew(async () =>
            {
                try
                {
                    while (true)
                    {
                        topToken.ThrowIfCancellationRequested();

                        try
                        {
                            using (var dbConnection = new SqlConnection(ConnectionString))
                            using (var command = new SqlCommand("marc.GetEvents", dbConnection))
                            {
                                await command.Connection.OpenAsync().ConfigureAwait(false);
                                command.CommandType = CommandType.StoredProcedure;
                                command.Parameters.AddWithValue("@fromId", lastEventId);

                                using (var reader = await command.ExecuteReaderAsync(topToken).ConfigureAwait(false))
                                {
                                    int received = lastEventId;
                                    while (await reader.ReadAsync(topToken).ConfigureAwait(false))
                                    {
                                        /// do stuff...
                                    }
                                    lastEventId = received;
                                }
                            }
                            await Task.Delay(PollIntervalMilliseconds, topToken).ConfigureAwait(false);
                        }
                        catch (OperationCanceledException)
                        {
                            throw;
                        }
                        catch (Exception ex)
                        {
                            if (ex is SqlException && topToken.IsCancellationRequested)
                            {
                                throw new OperationCanceledException("Operation cancelled by user", ex);
                            }

                            logger.Warn(ex, $"Exception on polling Codeks db. Waiting {delayOnSqlError}ms..."); // this is hit
                            _OnReaderEvent.OnError(ex);
                            await Task.Delay(delayOnSqlError, topToken).ConfigureAwait(false); // probably not executed
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    logger.Info("Listening task ended. Service is stopping?");
                }
                catch (Exception ex)
                {
                    logger.Error(ex, "General exception"); // falling here
                }
            }, TaskCreationOptions.LongRunning).ConfigureAwait(false);

今天我收到一份报告,说这项任务提前结束了。根据日志,第一个 catch

2018-08-01 17:42:08.6348 |警告|轮询代码数据库异常。等待5000毫秒。。。System.Data.SqlClient.SqlException(0x80131904):事务(进程ID 53)在与另一个进程的锁|通信缓冲区资源上死锁,并已被选为死锁牺牲品。重新运行事务。

但是,它并没有延迟,而是从循环中直接掉到外部 接住

2018-08-01 17:42:08.6488 | Error | Jantar.CodeksConnector | General exception System.Data.SqlClient.SqlException(0x80131904):事务(进程ID 53)在与另一个进程的锁|通信缓冲区资源上死锁,并被选为死锁受害者。重新运行事务。位于System.Data.SqlClient.SqlConnection.OneError(SqlException异常、布尔断开连接、操作 1 wrapCloseInAction) at System.Data.SqlClient.SqlInternalConnection.OnError(SqlException exception, Boolean breakConnection, Action 1(不作为) 位于System.Data.SqlClient.TdsParser.ThroweException和Warning(TdsParserStateObject StateObjectStateObj、布尔调用方连接锁、布尔异步关闭) 位于System.Data.SqlClient.SqlDataReader&燃气轮机;c___显示类189_0.b__0(任务t) 2 moreFunc, TaskCompletionSource 1个源,IDisposable objectToDispose) at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(任务任务) at System.Runtime.CompilerServices.TaskAwaiter.ValidateEnd(任务任务) 在Jantar.CodeksConnector;b_u18_u0>d、 MoveNext() ---来自引发异常的上一个位置的堆栈结束跟踪--- 在System.Responsive.PlatformServices.ExceptionServicesImpl.Rethrow(异常异常) at系统反应性短截线。<&燃气轮机;c、 <。cctor>b_uuu2_u1(例外情况除外) 在System.Reactive.AnonymousSafeObserver上 1.OnError(Exception error) at System.Reactive.Linq.ObservableImpl.SelectMany .OnError(异常错误) 在System.Reactive.Linq.ObservableImpl.Where 1._.OnError(Exception error) at System.Reactive.Linq.ObservableImpl.AsObservable 1. at System.Reactive.Observer 1.OnError(Exception error) at System.Reactive.Subjects.Subject

我没有主意了。。。

[更新:08.03]

Subject<T>.onError(ex) (我删除了它,因为它是一个bug)。这是一个双重错误,因为没有错误。我不知道在这种情况下,异常是rethrown,但只有在有订阅者的情况下才会出现,如果没有订阅者,则会被吞并。

1 回复 | 直到 6 年前

Enigmativity 6 年前

虽然这不是您问题的直接答案,但您已将其标记为“System.Reactive”,因此我想我可以(大致)向您展示代码的Rx解决方案。请记住,我不能准确地为 /// do stuff...

这是处方:

IObservable<string> query =
    from t in Observable.Interval(TimeSpan.FromMilliseconds(PollIntervalMilliseconds))
    from x in Observable.Using(
        () => new SqlConnection(ConnectionString),
        dbConnection =>
            Observable.Using(
                () =>
                {
                    var c = new SqlCommand("marc.GetEvents", dbConnection);
                    c.CommandType = CommandType.StoredProcedure;
                    c.Parameters.AddWithValue("@fromId", lastEventId);
                    return c;
                },
                command =>
                    from o in Observable.FromAsync(() => command.Connection.OpenAsync())
                    from reader in Observable.FromAsync(() => command.ExecuteReaderAsync(topToken))
                    let received = lastEventId
                    from r in Observable.FromAsync(() => reader.ReadAsync(topToken))
                    select reader.GetFieldValue<string>(0)))
    select x;