Последовательная обработка асинхронных задач
Предположим, что следующий синхронный код:
try
{
Foo();
Bar();
Fubar();
Console.WriteLine("All done");
}
catch(Exception e) // For illustration purposes only. Catch specific exceptions!
{
Console.WriteLine(e);
}
Теперь предположим, что все эти методы имеют аналог Async, и я должен использовать их по какой-то причине, поэтому простое перенос всего объекта в новую задачу не является вариантом.
Как я могу достичь такого же поведения?
Что я имею в виду с "тем же":
- Выполните обработчик для исключения, если вы его выбросите.
- Остановить выполнение следующих методов, если выбрано исключение.
Единственное, что я смог придумать, - ужасно:
var fooTask = FooAsync();
fooTask.ContinueWith(t => HandleError(t.Exception),
TaskContinuationOptions.OnlyOnFaulted);
fooTask.ContinueWith(
t =>
{
var barTask = BarAsync();
barTask.ContinueWith(t => HandleError(t.Exception),
TaskContinuationOptions.OnlyOnFaulted);
barTask.ContinueWith(
t =>
{
var fubarTask = FubarAsync();
fubarTask.ContinueWith(t => HandleError(t.Exception),
TaskContinuationOptions.OnlyOnFaulted);
fubarTask.ContinueWith(
t => Console.WriteLine("All done"),
TaskContinuationOptions.OnlyOnRanToCompletion);
},
TaskContinuationOptions.OnlyOnRanToCompletion);
},
TaskContinuationOptions.OnlyOnRanToCompletion);
Обратите внимание:
- Мне нужно решение, которое работает с .NET 4, поэтому
async/await
не может быть и речи. Однако, если он будет работать с async/await
, не стесняйтесь показать, как это сделать.
- Мне не нужно использовать TPL. Если это невозможно с помощью TPL, другой подход будет в порядке, возможно, с реактивными расширениями?
Ответы
Ответ 1
Вот как это работает с async
:
try
{
await FooAsync();
await BarAsync();
await FubarAsync();
Console.WriteLine("All done");
}
catch(Exception e) // For illustration purposes only. Catch specific exceptions!
{
Console.WriteLine(e);
}
Это будет работать с .NET 4.0, если вы установили пакет Stephen Toub Then
:
public static Task Then(this Task first, Func<Task> next)
{
var tcs = new TaskCompletionSource<object>();
first.ContinueWith(_ =>
{
if (first.IsFaulted) tcs.TrySetException(first.Exception.InnerExceptions);
else if (first.IsCanceled) tcs.TrySetCanceled();
else
{
try
{
next().ContinueWith(__ =>
{
if (t.IsFaulted) tcs.TrySetException(t.Exception.InnerExceptions);
else if (t.IsCanceled) tcs.TrySetCanceled();
else tcs.TrySetResult(null);
}, TaskContinuationOptions.ExecuteSynchronously);
}
catch (Exception exc) { tcs.TrySetException(exc); }
}
}, TaskContinuationOptions.ExecuteSynchronously);
return tcs.Task;
}
Вы можете использовать его как таковое:
var task = FooAsync().Then(() => BarAsync()).Then(() => FubarAsync());
task.ContinueWith(t =>
{
if (t.IsFaulted || t.IsCanceled)
{
var e = t.Exception.InnerException;
// exception handling
}
else
{
Console.WriteLine("All done");
}
}, TaskContinuationOptions.ExcecuteSynchronously);
Используя Rx, он будет выглядеть так (если у вас нет методов async
, уже выставленных как IObservable<Unit>
):
FooAsync().ToObservable()
.SelectMany(_ => BarAsync().ToObservable())
.SelectMany(_ => FubarAsync().ToObservable())
.Subscribe(_ => { Console.WriteLine("All done"); },
e => { Console.WriteLine(e); });
Я думаю. Я не мастер Rx, никоим образом.:)
Ответ 2
Как раз для полноты, как я реализую вспомогательный метод, предложенный Крисом Синклером:
public void RunSequential(Action onComplete, Action<Exception> errorHandler,
params Func<Task>[] actions)
{
RunSequential(onComplete, errorHandler,
actions.AsEnumerable().GetEnumerator());
}
public void RunSequential(Action onComplete, Action<Exception> errorHandler,
IEnumerator<Func<Task>> actions)
{
if(!actions.MoveNext())
{
onComplete();
return;
}
var task = actions.Current();
task.ContinueWith(t => errorHandler(t.Exception),
TaskContinuationOptions.OnlyOnFaulted);
task.ContinueWith(t => RunSequential(onComplete, errorHandler, actions),
TaskContinuationOptions.OnlyOnRanToCompletion);
}
Это гарантирует, что каждая последующая задача запрашивается только при успешном завершении предыдущей.
Предполагается, что Func<Task>
возвращает уже запущенную задачу.
Ответ 3
То, что у вас здесь, по существу есть ForEachAsync
. Вы хотите последовательно запускать каждый асинхронный элемент, но с некоторой поддержкой обработки ошибок. Вот одна из таких реализаций:
public static Task ForEachAsync(IEnumerable<Func<Task>> tasks)
{
var tcs = new TaskCompletionSource<bool>();
Task currentTask = Task.FromResult(false);
foreach (Func<Task> function in tasks)
{
currentTask.ContinueWith(t => tcs.TrySetException(t.Exception.InnerExceptions)
, TaskContinuationOptions.OnlyOnFaulted);
currentTask.ContinueWith(t => tcs.TrySetCanceled()
, TaskContinuationOptions.OnlyOnCanceled);
Task<Task> continuation = currentTask.ContinueWith(t => function()
, TaskContinuationOptions.OnlyOnRanToCompletion);
currentTask = continuation.Unwrap();
}
currentTask.ContinueWith(t => tcs.TrySetException(t.Exception.InnerExceptions)
, TaskContinuationOptions.OnlyOnFaulted);
currentTask.ContinueWith(t => tcs.TrySetCanceled()
, TaskContinuationOptions.OnlyOnCanceled);
currentTask.ContinueWith(t => tcs.TrySetResult(true)
, TaskContinuationOptions.OnlyOnRanToCompletion);
return tcs.Task;
}
Я добавил в поддержку отмененных задач, просто для того, чтобы быть более общим и потому, что так мало дел.
Он добавляет каждую задачу в качестве продолжения предыдущей задачи, и по всей линии она гарантирует, что любые исключения приводят к установке конечного исключения задачи.
Вот пример использования:
public static Task FooAsync()
{
Console.WriteLine("Started Foo");
return Task.Delay(1000)
.ContinueWith(t => Console.WriteLine("Finished Foo"));
}
public static Task BarAsync()
{
return Task.Factory.StartNew(() => { throw new Exception(); });
}
private static void Main(string[] args)
{
List<Func<Task>> list = new List<Func<Task>>();
list.Add(() => FooAsync());
list.Add(() => FooAsync());
list.Add(() => FooAsync());
list.Add(() => FooAsync());
list.Add(() => BarAsync());
Task task = ForEachAsync(list);
task.ContinueWith(t => Console.WriteLine(t.Exception.ToString())
, TaskContinuationOptions.OnlyOnFaulted);
task.ContinueWith(t => Console.WriteLine("Done!")
, TaskContinuationOptions.OnlyOnRanToCompletion);
}
Ответ 4
Вы должны иметь возможность создать метод для объединения двух задач и только запустить второй, если первый преуспеет.
public static Task Then(this Task parent, Task next)
{
TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
parent.ContinueWith(pt =>
{
if (pt.IsFaulted)
{
tcs.SetException(pt.Exception.InnerException);
}
else
{
next.ContinueWith(nt =>
{
if (nt.IsFaulted)
{
tcs.SetException(nt.Exception.InnerException);
}
else { tcs.SetResult(null); }
});
next.Start();
}
});
return tcs.Task;
}
вы можете объединить задачи вместе:
Task outer = FooAsync()
.Then(BarAsync())
.Then(FubarAsync());
outer.ContinueWith(t => {
if(t.IsFaulted) {
//handle exception
}
});
Если ваши задачи запускаются немедленно, вы можете просто их обернуть в Func
:
public static Task Then(this Task parent, Func<Task> nextFunc)
{
TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
parent.ContinueWith(pt =>
{
if (pt.IsFaulted)
{
tcs.SetException(pt.Exception.InnerException);
}
else
{
Task next = nextFunc();
next.ContinueWith(nt =>
{
if (nt.IsFaulted)
{
tcs.SetException(nt.Exception.InnerException);
}
else { tcs.SetResult(null); }
});
}
});
return tcs.Task;
}
Ответ 5
Теперь, я не очень использовал TPL, так что это просто удар в темноте. И учитывая то, что сказал @Servy, возможно, это не будет выполняться полностью асинхронно. Но я решил, что я опубликую его, и если это будет не по себе, вы можете опустить меня в забвение или я могу удалить его (или мы можем исправить то, что нужно для исправления)
public void RunAsync(Action onComplete, Action<Exception> errorHandler, params Action[] actions)
{
if (actions.Length == 0)
{
//what to do when no actions/tasks provided?
onComplete();
return;
}
List<Task> tasks = new List<Task>(actions.Length);
foreach(var action in actions)
{
Task task = new Task(action);
task.ContinueWith(t => errorHandler(t.Exception), TaskContinuationOptions.OnlyOnFaulted);
tasks.Add(task);
}
//last task calls onComplete
tasks[actions.Length - 1].ContinueWith(t => onComplete(), TaskContinuationOptions.OnlyOnRanToCompletion);
//wire all tasks to execute the next one, except of course, the last task
for (int i = 0; i <= actions.Length - 2; i++)
{
var nextTask = tasks[i + 1];
tasks[i].ContinueWith(t => nextTask.Start(), TaskContinuationOptions.OnlyOnRanToCompletion);
}
tasks[0].Start();
}
И это будет использовать как:
RunAsync(() => Console.WriteLine("All done"),
ex => Console.WriteLine(ex),
Foo,
Bar,
Fubar);
Мысли? Downvotes?:)
(я определенно предпочитаю async/await, хотя)
ИЗМЕНИТЬ: на основании ваших комментариев, чтобы принять Func<Task>
, будет ли это правильной реализацией?
public void RunAsync(Action onComplete, Action<Exception> errorHandler, params Func<Task>[] actions)
{
if (actions.Length == 0)
{
//what to do when no actions/tasks provided?
onComplete();
return;
}
List<Task> tasks = new List<Task>(actions.Length);
foreach (var action in actions)
{
Func<Task> nextActionFunc = action;
Task task = new Task(() =>
{
var nextTask = nextActionFunc();
nextTask.ContinueWith(t => errorHandler(t.Exception), TaskContinuationOptions.OnlyOnFaulted);
nextTask.Start();
});
tasks.Add(task);
}
//last task calls onComplete
tasks[actions.Length - 1].ContinueWith(t => onComplete(), TaskContinuationOptions.OnlyOnRanToCompletion);
//wire all tasks to execute the next one, except of course, the last task
for (int i = 0; i <= actions.Length - 2; i++)
{
var nextTask = tasks[i + 1];
tasks[i].ContinueWith(t => nextTask.Start(), TaskContinuationOptions.OnlyOnRanToCompletion);
}
tasks[0].Start();
}