使用 ActionBlock 和 SendAsync() 构建您所要求的内容相当简单，类似于：
public static ITargetBlock<T> CreateGuaranteedBroadcastBlock<T>(
IEnumerable<ITargetBlock<T>> targets)
{
var targetsList = targets.ToList();
return new ActionBlock<T>(
async item =>
{
foreach (var target in targetsList)
{
await target.SendAsync(item);
}
}, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });
}
这是最基本的版本，但将其扩展以支持可变目标列表、传播完成或克隆函数应该很容易。
GuaranteedDeliveryBroadcastBlock类是内置BroadcastBlock的（几乎）完全替代品。随时支持链接和取消链接目标。public class GuaranteedDeliveryBroadcastBlock<T> :
ITargetBlock<T>, ISourceBlock<T>, IPropagatorBlock<T, T>
{
private class Subscription
{
public readonly ITargetBlock<T> Target;
public readonly bool PropagateCompletion;
public readonly CancellationTokenSource CancellationSource;
public Subscription(ITargetBlock<T> target,
bool propagateCompletion,
CancellationTokenSource cancellationSource)
{
Target = target;
PropagateCompletion = propagateCompletion;
CancellationSource = cancellationSource;
}
}
private readonly object _locker = new object();
private readonly Func<T, T> _cloningFunction;
private readonly CancellationToken _cancellationToken;
private readonly ITargetBlock<T> _actionBlock;
private readonly List<Subscription> _subscriptions = new List<Subscription>();
private readonly Task _completion;
private CancellationTokenSource _faultCTS
= new CancellationTokenSource(); // Is nullified on completion
public GuaranteedDeliveryBroadcastBlock(Func<T, T> cloningFunction,
DataflowBlockOptions dataflowBlockOptions = null)
{
_cloningFunction = cloningFunction
?? throw new ArgumentNullException(nameof(cloningFunction));
dataflowBlockOptions ??= new DataflowBlockOptions();
_cancellationToken = dataflowBlockOptions.CancellationToken;
_actionBlock = new ActionBlock<T>(async item =>
{
Task sendAsyncToAll;
lock (_locker)
{
var allSendAsyncTasks = _subscriptions
.Select(sub => sub.Target.SendAsync(
_cloningFunction(item), sub.CancellationSource.Token));
sendAsyncToAll = Task.WhenAll(allSendAsyncTasks);
}
await sendAsyncToAll;
}, new ExecutionDataflowBlockOptions()
{
CancellationToken = dataflowBlockOptions.CancellationToken,
BoundedCapacity = dataflowBlockOptions.BoundedCapacity,
MaxMessagesPerTask = dataflowBlockOptions.MaxMessagesPerTask,
TaskScheduler = dataflowBlockOptions.TaskScheduler,
});
var afterCompletion = _actionBlock.Completion.ContinueWith(t =>
{
lock (_locker)
{
// PropagateCompletion
foreach (var subscription in _subscriptions)
{
if (subscription.PropagateCompletion)
{
if (t.IsFaulted)
subscription.Target.Fault(t.Exception);
else
subscription.Target.Complete();
}
}
// Cleanup
foreach (var subscription in _subscriptions)
{
subscription.CancellationSource.Dispose();
}
_subscriptions.Clear();
_faultCTS.Dispose();
_faultCTS = null; // Prevent future subscriptions to occur
}
}, TaskScheduler.Default);
// Ensure that any exception in the continuation will be surfaced
_completion = Task.WhenAll(_actionBlock.Completion, afterCompletion);
}
public Task Completion => _completion;
public void Complete() => _actionBlock.Complete();
void IDataflowBlock.Fault(Exception ex)
{
_actionBlock.Fault(ex);
lock (_locker) _faultCTS?.Cancel();
}
public IDisposable LinkTo(ITargetBlock<T> target,
DataflowLinkOptions linkOptions)
{
if (linkOptions.MaxMessages != DataflowBlockOptions.Unbounded)
throw new NotSupportedException();
Subscription subscription;
lock (_locker)
{
if (_faultCTS == null) return new Unlinker(null); // Has completed
var cancellationSource = CancellationTokenSource
.CreateLinkedTokenSource(_cancellationToken, _faultCTS.Token);
subscription = new Subscription(target,
linkOptions.PropagateCompletion, cancellationSource);
_subscriptions.Add(subscription);
}
return new Unlinker(() =>
{
lock (_locker)
{
// The subscription may have already been removed
if (_subscriptions.Remove(subscription))
{
subscription.CancellationSource.Cancel();
subscription.CancellationSource.Dispose();
}
}
});
}
private class Unlinker : IDisposable
{
private readonly Action _action;
public Unlinker(Action disposeAction) => _action = disposeAction;
void IDisposable.Dispose() => _action?.Invoke();
}
DataflowMessageStatus ITargetBlock<T>.OfferMessage(
DataflowMessageHeader messageHeader, T messageValue,
ISourceBlock<T> source, bool consumeToAccept)
{
return _actionBlock.OfferMessage(messageHeader, messageValue, source,
consumeToAccept);
}
T ISourceBlock<T>.ConsumeMessage(DataflowMessageHeader messageHeader,
ITargetBlock<T> target, out bool messageConsumed)
=> throw new NotSupportedException();
bool ISourceBlock<T>.ReserveMessage(DataflowMessageHeader messageHeader,
ITargetBlock<T> target)
=> throw new NotSupportedException();
void ISourceBlock<T>.ReleaseReservation(DataflowMessageHeader messageHeader,
ITargetBlock<T> target)
=> throw new NotSupportedException();
}
缺少的功能: 未实现 IReceivableSourceBlock<T> 接口，且不支持使用 MaxMessages 选项进行链接。
此类是线程安全的。
GuaranteedDeliveryBroadcastBlock 基于 svick 的想法，即拥有一个中央的 ActionBlock 将消息分发到各个目标。我只是将这个功能打包成了一个漂亮的 IPropagatorBlock 实现，以便更容易地使用它。lock 是必需的，以允许并发链接和取消链接目标。如果没有一些形式的同步机制，我无法在结尾声称“此类是线程安全的”。而锁是最简单、最容易理解的工具。 - Theodor Zoulias
BroadcastBlock。你能否在某个地方发布演示问题的示例代码？ - svick