我有一个TPL数据流操作块，我用它来接收相机的触发消息，然后进行一些处理。 如果处理任务引发异常，则操作块将进入故障状态。 我想向我的 UI 发送错误消息，并向操作块发送重置消息，以便它可以继续处理传入的触发消息。 有没有办法将动作块返回到就绪状态(清除故障)？
好奇的代码：
using System.Threading.Tasks.Dataflow;
namespace Anonymous
{
/// <summary>
/// Provides a messaging system between objects that inherit from Actor
/// </summary>
public abstract class Actor
{
//The Actor uses an ActionBlock from the DataFlow library. An ActionBlock has an input queue you can
// post messages to and an action that will be invoked for each received message.
//The ActionBlock handles all of the threading issues internally so that we don't need to deal with
// threads or tasks. Thread-safety comes from the fact that ActionBlocks are serialized by default.
// If you send two messages to it at the same time it will buffer the second message until the first
// has been processed.
private ActionBlock<Message> _action;
...Properties omitted for brevity...
public Actor(string name, int id)
{
_name = name;
_id = id;
CreateActionBlock();
}
private void CreateActionBlock()
{
// We create an action that will convert the actor and the message to dynamic objects
// and then call the HandleMessage method. This means that the runtime will look up
// a method called ‘HandleMessage’ with a parameter of the message type and call it.
// in TPL Dataflow if an exception goes unhandled during the processing of a message,
// (HandleMessage) the exception will fault the block’s Completion task.
//Dynamic objects expose members such as properties and methods at run time, instead
// of at compile time. This enables you to create objects to work with structures that
// do not match a static type or format.
_action = new ActionBlock<Message>(message =>
{
dynamic self = this;
dynamic msg = message;
self.HandleMessage(msg); //implement HandleMessage in the derived class
}, new ExecutionDataflowBlockOptions
{
MaxDegreeOfParallelism = 1 // This specifies a maximum degree of parallelism of 1.
// This causes the dataflow block to process messages serially.
});
}
/// <summary>
/// Send a message to an internal ActionBlock for processing
/// </summary>
/// <param name="message"></param>
public async void SendMessage(Message message)
{
if (message.Source == null)
throw new Exception("Message source cannot be null.");
try
{
_action.Post(message);
await _action.Completion;
message = null;
//in TPL Dataflow if an exception goes unhandled during the processing of a message,
// the exception will fault the block’s Completion task.
}
catch(Exception ex)
{
_action.Completion.Dispose();
//throw new Exception("ActionBlock for " + _name + " failed.", ex);
Trace.WriteLine("ActionBlock for " + _name + " failed." + ExceptionExtensions.GetFullMessage(ex));
if (_action.Completion.IsFaulted)
{
_isFaulted = true;
_faultReason = _name + " ActionBlock encountered an exception while processing task: " + ex.ToString();
FaultMessage msg = new FaultMessage { Source = _name, FaultReason = _faultReason, IsFaulted = _isFaulted };
OnFaulted(msg);
CreateActionBlock();
}
}
}
public event EventHandler<FaultMessageEventArgs> Faulted;
public void OnFaulted(FaultMessage message)
{
Faulted?.Invoke(this, new FaultMessageEventArgs { Message = message.Copy() });
message = null;
}
/// <summary>
/// Use to await the message processing result
/// </summary>
public Task Completion
{
get
{
_action.Complete();
return _action.Completion;
}
}
}
}
操作块中未处理的异常类似于应用程序中未处理的异常。别这样。适当地处理异常。
在最简单的情况下，记录它或在块的委托中执行某些操作。在更复杂的方案中，您可以使用转换块而不是操作块，并将成功或失败消息发送到下游块。
不过，您发布的代码存在一些关键问题。数据流块不是代理，代理也不是数据流块。当然，您可以使用一个来构建另一个，但它们代表不同的范式。在这种情况下，您的Actor会模拟ActionBlock自己的 API，但存在多个错误。
例如，您不需要创建SendAsync，块已经有一个。发送消息后，不应完成阻止。您将无法处理任何其他消息。仅当您确实不想再使用动作块时才调用Complete()。您无需将DOP 设置为 1，这是默认值。
您可以设置数据流块的边界，以便它一次只接受 10 条消息。否则，所有消息都将被缓冲，直到块找到处理它们的机会
您可以将所有这些代码替换为以下内容：
void MyMethod(MyMessage message)
{
try
{
//...
}
catch(Exception exc)
{
//ToString logs the *complete exception, no need for anything more
_log.Error(exc.ToString());
}
}
var blockOptions new ExecutionDataflowBlockOptions {
BoundedCapacity=10,
NameFormat="Block for MyMessage {0} {1}"
};
var block=new ActionBlock<MyMessage>(MyMethod,blockOptions);
for(int i=0;i<10000;i++)
{
//Will await if more than 10 messages are waiting
await block.SendAsync(new MyMessage(i);
}
block.Complete();
//Await until all leftover messages are processed
await block.Completion;
请注意对Exception.ToString()的调用。这将生成一个包含所有异常信息(包括调用堆栈)的字符串。
NameFormat允许您为块指定名称模板，运行时可以使用块的内部名称和任务 ID 填充该模板。