c# - 类似于文件系统日志记录或容错日志记录的过程来管理任务？

Question

开发一个 c# 应用程序，该应用程序将根据用户上传的文件运行一系列任务；这可能需要几秒钟到几天的时间。我计划实现某种日志系统以在流程中断时恢复工作。这个中断日志/恢复日志叫什么？我在哪里可以了解有关现有实现的更多信息？

编辑:

到目前为止我发现的有用的东西:

A journalled file system would use the same basic procedure, with a few additional steps. Something like:

• Journal entry: Moving file from A to B
  • Physically copy old file to new location
  • Update directory entry on new drive
  • Remove directory entry from old drive
  • Free space on old drive
• Journal entry: Done moving file from A to B

来自 https://serverfault.com/questions/173176/what-is-a-journaling-file-system

---

Checkpointing: The process of writing journaled metadata and data to their fixed-locations is known as checkpointing. Checkpointing is triggered when various thresholds are crossed, e.g., when file system buffer space is low, when there is little free space left in the journal, or when a timer expires.

Crash Recovery: Crash recovery is straightforward in ext3 (as it is in many journaling file systems); a basic form of redo logging is used. Because new updates (whether to data or just metadata) are written to the log, the process of restoring in-place file system structures is easy. During recovery, the file system scans the log for committed complete transactions; incomplete transactions are discarded. Each update in a completed transaction is simply replayed into the fixed-place ext2 structures.

来自 http://research.cs.wisc.edu/adsl/Publications/sba-usenix05.pdf (第 5 页)

---

编辑 2:

我觉得我对容错的了解并不比我第一次发布这个问题时多多少，但这里是我实现内容的概述。

作业管理器的主要尝试是从文件加载任何现有的已保存状态，或创建一个新的空管理器。

public static void Main(string[] args)
{
    try
    {
            _jxman = JobManager<Job>.Load(Properties.Settings.Default.JobJournalFilename);
    }
    catch
    {
            _jxman = new JobManager<Job>(Properties.Settings.Default.JobJournalFilename);
    }
    ...
    _jxman.Start();
    ...
}

JobManager 类看起来像这样

public sealed class JobManager<T> : WorkGroupBase<JobBase>, IWorkSerializable where T : IJob
{
        #region Fields

        /// <summary>
        /// Hash goes here in file
        /// </summary>
        private const string _hashHeading = "SHA-256";

        /// <summary>
        /// Flag to know whether to update the journal
        /// </summary>
        private bool _isDirty = false;

        /// <summary>
        /// Last time the journal was written to disk
        /// </summary>
        private DateTime _lastSaveTime = DateTime.MinValue;

        /// <summary>
        /// Minimum time to wait before writing journal to disk again
        /// </summary>
        private TimeSpan _minTimeToSave = new TimeSpan(0,0,60);

        /// <summary>
        /// Threading object for lock
        /// </summary>
        private object _lock = new object();

        /// <summary>
        /// Thread to monitor status
        /// </summary>
        private Thread _watchDirtyFlag;

        #endregion

        #region Properties

        /// <summary>
        /// journal file to track changes
        /// </summary>
        public string Filename
        {
                get;
                private set;
        }

        #endregion

        #region Constructors

        /// <summary>
        /// default constructor
        /// </summary>
        /// <param name="filename">Path to filename to write journal file</param>
        public JobManager(string filename) : base()
        {
                ConstructorHelper();

                Filename = filename;
        }

        /// <summary>
        /// Parses XML element to recreate the item
        /// </summary>
        /// <param name="xe">XML element used to create object</param>
        public JobManager(XElement xe)
                : base(xe)
        {
                // Checksum validation before doing anything else.
                // Will throw exception on failure.
                ValidateChecksum(xe);

                ConstructorHelper();

                string myName = "JobManager";

                XElement myself;
                try
                {
                        myself = xe.DescendantsAndSelf(myName).First();
                }
                catch
                {
                        throw new ArgumentException("Attempting to instantiate object, but no relevant information was found in the XML element");
                }

                Filename = myself.FirstElementValue("Filename");

                // Load up all the jobs
                XElement[] wq = myself.Descendants("WorkQueue").Elements().ToArray();

                foreach (XElement x in wq)
                {
                        try
                        {
                                IJob blarg = (IJob)Activator.CreateInstance(typeof(T), x);
                                if (blarg != null)
                                        WorkQueue.Enqueue((JobBase)blarg);
                        }
                        catch
                        { }
                }
        }

        /// <summary>
        /// Helper for common constructing 
        /// </summary>
        private void ConstructorHelper()
        {
                // need to wait for the base constructor to finish before attempting to
                // hook events there
                base.QueueChanged += new EventHandler(JobManager_QueueChanged);
                base.HookQueueChangedEvents();

                _watchDirtyFlag = new Thread(WatchDirtyFlag);
                _watchDirtyFlag.Start();
        }

        #endregion

        #region Methods

        /// <summary>
        /// Saves the state of the JobManager to Filename using XML
        /// </summary>
        public void Save()
        {
                TextWriter writer = null;
                try
                {
                        writer = new StreamWriter(Filename);
                        writer.Write(this.ToXElement());
                }
                catch (Exception ex)
                {
                        throw ex;
                }
                finally
                {
                        writer.Close();
                }
        }

        /// <summary>
        /// Loads the filename and attempts to parse it as XML to 
        /// create a JobManager. Pass the type of job to manage.
        /// </summary>
        /// <param name="filename">File storing the JobManager as XML</param>
        /// <returns>JobManager with values loaded from file</returns>
        public static JobManager<T> Load(string filename)
        {
                if (filename == "")
                        throw new ArgumentException("Can not load JobManager: Filename not set");

                TextReader reader = null;
                string text;
                try
                {
                        reader = new StreamReader(filename);
                        text = reader.ReadToEnd();
                }
                catch (Exception ex)
                {
                        throw ex;
                }
                finally
                {
                        reader.Close();
                }

                XElement loadFrom = null;
                try
                {
                        loadFrom = XElement.Parse(text);
                }
                catch //(Exception ex)
                {
                        //throw ex;
                        loadFrom = new XElement("empty");
                }

                JobManager<T> output = new JobManager<T>(loadFrom);
                output.Filename = filename;
                return output;
        }

        /// <summary>
        /// Converts the item to an XML element
        /// </summary>
        /// <returns></returns>
        new public XElement ToXElement()
        {
                XElement bxe = base.ToXElement();

                //string myName = this.GetType().Name;
                string myName = "JobManager";

                XElement wq = new XElement("WorkQueue");
                foreach (IWorkSerializable t in WorkQueue.ToArray())
                {
                        XElement addee = t.ToXElement();

                        wq.Add(addee);
                }

                bxe.Add(wq);

                XElement xe = new XElement(myName,
                        bxe,
                        new XElement("Filename", Filename)
                        );

                xe.Add(
                        new XElement(_hashHeading, Generic.ComputeSha256Hash(xe.ToString()))
                        );

                return xe;
        }

        /// <summary>
        /// Validates the checksum for the current xelement. Throws exceptions on failure
        /// </summary>
        /// <param name="xe">XML tree of the itme to validate</param>
        private void ValidateChecksum(XElement xe)
        {
                XElement checksum;
                try
                {
                        checksum = xe.DescendantsAndSelf(_hashHeading).First();
                }
                catch (Exception ex)
                {
                        throw new Exception("Unable to find checksum node", ex);
                }

                XElement withoutChecksum = new XElement(xe);
                withoutChecksum.Elements(_hashHeading).Remove();

                string computedChecksum = Generic.ComputeSha256Hash(withoutChecksum.ToString());

                if (computedChecksum != checksum.Value)
                        throw new Exception("Checksum from XML element and checksum from contents of XML element do not match: \n" + xe.Value);


        }

        /// <summary>
        /// This thread will watch the dirty flag, which is set everytime the 
        /// queues are changed. Every _minTimeToSave the flag is checked, and
        /// if the flag is set, Save() is called.
        /// </summary>
        private void WatchDirtyFlag()
        {
                while (true)
                {
                        // sleep until there's something to update
                        while (_isDirty == false)
                        {
                                Thread.Sleep(_minTimeToSave);
                        }

                        // but don't update too frequently
                        if (DateTime.Now.Subtract(_lastSaveTime) > _minTimeToSave)
                        {
                                // save first ...
                                this.Save();
                                // then update items ...
                                _lastSaveTime = DateTime.Now;
                                lock (_lock)
                                {
                                        _isDirty = false;
                                }
                        }
                }
        }

        #endregion

        #region Event Handlers

        /// <summary>
        /// updates flag when any underlying queue changes
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        void JobManager_QueueChanged(object sender, EventArgs e)
        {
                lock (_lock)
                {
                        _isDirty = true;
                }
        }

        #endregion
}

注意事项:

• 这是不完整的代码，以防有人试图复制它(缺少基类和东西)
• 常规(二进制)和 XML 序列化从来没有完全正确地工作过，所以我实现了将对象保存为 XML 的自定义序列化。这是 ToXElement() 方法，以及采用 XElement 参数的构造函数。
• 在序列化的顶层 (JobManager) 中包含一个校验和 (SHA-256)。从 XElement 实例化新对象时，将保存的序列化对象的校验和与文件中的校验和进行比较。
• 有一个静态方法 .Load(file)，它通过读取文件并尝试反序列化内容来返回一个新的 JobManager 对象。
• 此处未显示自定义的 ConcurrentQueue 类。这是 MSDN ConcurrentQueue 的包装器类，但添加了一个事件以在队列更改时发出通知。
• 这个 JobManager 类实现了一个带有上述 ConcurrentQueue 的基类；这些队列更改事件 Hook 在构造函数助手中
• 当事件触发时，JobManager 设置一个标志 _isDirty
• JobManager 在实例化时启动一个线程，该线程监视 _isDirty 标志。大部分时间都花在 sleep 上，但如果至少 _minTimeToSave 已经过去，JobManager 的内容将序列化到磁盘。这应该可以防止 JobManager 过于频繁地写入磁盘。

Unresolved 问题:

• 线程真的是观察 _isDirty 标志的正确解决方案吗？
• JobManager(单线程)管理包含任务的作业(一次一个，但不同的线程)；在序列化时没有类到基类的同步来锁定状态
• 旧的已完成作业被序列化到磁盘，然后重新加载

Answer 1

我不确定是否有用于此目的的现有系统，但 Serialization是实现这种排序实现的关键。您只需设计您的对象以支持序列化。

• 在任务中断的情况下，您可以以任何格式(二进制或XML)保存或序列化对象的状态) 在文件系统上。

• 为了恢复任务，您只需反序列化对象，然后您就可以恢复业务了。

要了解有关序列化的更多信息，请使用以下引用资料:

1. What is [Serializable] and when should I use it?
2. .NET Serialization (using BinaryFormater, SoapFormatter and XmlSerializer)