1

我有一个 TPL 数据流操作块，用于接收相机的触发消息，然后进行一些处理。如果处理任务抛出异常，则 ActionBlock 进入故障状态。我想向我的 UI 发送一条错误消息并向 ActionBlock 发送一条重置消息，以便它可以继续处理传入的触发消息。有没有办法将 ActionBlock 返回到就绪状态（清除故障）？

好奇的代码：

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;
            }
        }
    }
}
4

1 回答 1

2

ActionBlock 中未处理的异常类似于应用程序中的未处理异常。不要这样做。适当处理异常。

在最简单的情况下，记录它或在块的委托中做一些事情。在更复杂的场景中，您可以使用 TransformBlock 代替 ActionBlock 并将成功或失败消息发送到下游块。

您发布的代码虽然有一些关键问题。数据流块不是代理，代理也不是数据流块。当然，您可以使用一个来构建另一个，但它们代表不同的范例。在这种情况下，您会Actor模拟ActionBlock带有几个错误的自己的 API。

例如，您不需要创建一个SendAsync, 块已经有一个。发送消息后，您不应完成阻止。您将无法处理任何其他消息。仅Complete()在您真的不想再使用 ActionBlock 时才调用。您不需要将 DOP 设置为 1，这是默认值。

您可以为 DataflowBlock 设置边界，以便它一次只接受例如 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 填充该名称模板。

于 2018-04-05T10:15:05.343 回答