Streamwriter、StringBuilder 和并行循环

Question

抱歉，代码量很大，我无法用 less 来解释这一点。基本上我正在尝试从许多任务中写入文件。 你们能告诉我我做错了什么吗？ _streamWriter.WriteLine() 抛出 ArgumentOutOfRangeException。

class Program
{
    private static LogBuilder _log = new LogBuilder();
    static void Main(string[] args)
    {
        var acts = new List<Func<string>>();
        var rnd = new Random();
        for (int i = 0; i < 10000; i++)
        {
            acts.Add(() =>
            {
                var delay = rnd.Next(300);
                Thread.Sleep(delay);
                return "act that that lasted "+delay;
            });
        }

        Parallel.ForEach(acts, act =>
        {
            _log.Log.AppendLine(act.Invoke());
            _log.Write();
        });
    }
}

public class LogBuilder : IDisposable
{
    public StringBuilder Log = new StringBuilder();
    private FileStream _fileStream;
    private StreamWriter _streamWriter;

    public LogBuilder()
    {
        _fileStream = new FileStream("log.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.ReadWrite);
        _streamWriter = new StreamWriter(_fileStream) { AutoFlush = true };
    }
    public void Write()
    {
        lock (Log)
        {
            if (Log.Length <= 0) return;
            _streamWriter.WriteLine(Log.ToString()); //throws here. Although Log.Length is greater than zero
            Log.Clear();
        }
    }

    public void Dispose()
    {
        _streamWriter.Close(); _streamWriter.Dispose(); _fileStream.Close(); fileStream.Dispose();
    }
}

Answer 1

这不是StringBuilder 中的错误，而是您的代码中的错误。您在后续答案中显示的修改（将Log.String 替换为一次提取一个字符的循环）并不能解决它。它不会再抛出异常，但也不会正常工作。

问题是您在多线程代码中的两个地方使用StringBuilder，其中一个不尝试锁定它，这意味着读取可以在一个线程上同时发生在另一个线程上。特别是，问题在于这一行：

_log.Log.AppendLine(act.Invoke()); 

您正在您的Parallel.ForEach 中执行此操作。您在这里没有进行任何同步尝试，即使这将同时在多个线程上运行。所以你有两个问题：

1. 对AppendLine 的多次调用可能在多个线程上同时进行
2. 一个线程可能会尝试调用Log.ToString，同时一个或多个其他线程正在调用AppendLine

您一次只能读取一次，因为您使用 lock 关键字来同步这些内容。问题是您在调用AppendLine 时并未获得相同的锁。

您的“修复”并不是真正的修复。你成功的只是让问题更难被发现。它现在只会以不同的和更微妙的方式出错。例如，我假设您的Write 方法在您的for 循环完成其最终迭代后仍继续调用Log.Clear。在完成最后一次迭代和调用 Log.Clear 之间，有可能其他线程会再次调用 AppendLine，因为对 AppendLine 的调用没有同步。

结果是你有时会错过一些东西。代码会将内容写入字符串构建器，然后将其清除而无需写入流编写器。

此外，并发AppendLine 调用很有可能导致问题。如果你幸运的话，它们会不时崩溃。 （这很好，因为它清楚地表明你有一个问题需要解决。）如果你不走运，你会不时得到数据损坏 - 两个线程可能最终写入StringBuilder 中的同一个地方结果要么一团糟，要么完全丢失数据。

同样，这不是StringBuilder 中的错误。它并非旨在支持从多个线程同时使用。您的工作是确保一次只有一个线程对StringBuilder 的任何特定实例执行任何操作。正如该类的文档所述，“不保证任何实例成员都是线程安全的。”

显然，您不想在调用 act.Invoke() 时持有锁，因为这可能正是您想要并行化的工作。所以我猜这样的事情可能会更好：

string result = act();
lock(_log.Log)
{
    _log.Log.AppendLine(result);
}

但是，如果我把它留在那里，我就不会真正帮助你，因为这对我来说看起来很不对。

如果您发现自己将某个字段锁定在其他人的对象中，则表明您的代码存在设计问题。修改设计可能更有意义，以便 LogBuilder.Write 方法接受字符串。老实说，我什至不确定您为什么在这里使用StringBuilder，因为您似乎只是将它用作您立即写入流写入器的字符串的保存区域。你希望StringBuilder 在这里添加什么？以下会更简单，并且似乎不会丢失任何东西（除了原始的并发错误）：

public class LogBuilder : IDisposable
{
    private readonly object _lock = new object();
    private FileStream _fileStream;
    private StreamWriter _streamWriter;

    public LogBuilder()
    {
        _fileStream = new FileStream("log.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.ReadWrite);
        _streamWriter = new StreamWriter(_fileStream) { AutoFlush = true };
    }
    public void Write(string logLine)
    {
        lock (_lock)
        {
            _streamWriter.WriteLine(logLine);
        }
    }

    public void Dispose()
    {
        _streamWriter.Dispose(); fileStream.Dispose();
    }
}    

Answer 2

我认为原因是因为您正在访问 Parellel 括号中的 stringBuilder

_log.Log.AppendLine(act.Invoke());
_log.Write();

并在 LogBuilder 内部执行 lock() 以禁止在 stringBuidler 上分配内存。您正在更改流写入器以处理每个字符中的日志，因此将提供并行进程来解锁对 stringBuilder 的内存分配。

将并行过程分成不同的动作可能会减少问题

Parallel.ForEach(acts, act =>
{
    _log.Write(act.Invoke());
});

在 LogBuilder 类中

private readonly object _lock = new object();

public void Write(string logLines)
{
    lock (_lock)
    {
        //_wr.WriteLine(logLines);
        Console.WriteLine(logLines);
    }
}

Answer 3

另一种方法是使用TextWriter.Synchronized 包装StreamWriter。

void Main(string[] args)
{
    var rnd = new Random();
    var writer = new StreamWriter(@"C:\temp\foo.txt");
    var syncedWriter = TextWriter.Synchronized(writer);

    var tasks = new List<Func<string>>();
    for (int i = 0; i < 1000; i++)
    {
        int local_i = i; // get a local value, not closure-reference to i
        tasks.Add(() =>
        {
            var delay = rnd.Next(5);
            Thread.Sleep(delay);
            return local_i.ToString() + " act that that lasted " + delay.ToString();
        });
    }
    Parallel.ForEach(tasks, task =>
    {
        var value = task();
        syncedWriter.WriteLine(value);
    });

    writer.Dispose();
}

这里是一些同步助手类 http://referencesource.microsoft.com/#q=Synchronized

System.Collections

• 静态 ArrayList 同步（ArrayList 列表）
• 静态 IList 同步（IList 列表）
• 静态哈希表同步（哈希表）
• 静态队列同步（队列队列）
• 静态SortedList Synchronized(SortedList list)
• 静态堆栈同步（堆栈堆栈）

System.Collections.Generic

• 静态 IList 同步（列表列表）

System.IO

• 静态流同步（流流）
• 静态TextReader Synchronized(TextReader reader)
• 静态 TextWriter Synchronized(TextWriter writer)

System.Text.RegularExpressions

• 静态匹配同步（匹配内部）
• 静态组同步（组内）

Answer 4

看来这不是并行性的问题。这是 StringBuilder 的问题。

我已经换了：

_streamWriter.WriteLine(Log.ToString());

与：

for (int i = 0; i < Log.Length; i++)
{
    _streamWriter.Write(Log[i]);
}

它奏效了。 供日后参考：http://msdn.microsoft.com/en-us/library/system.text.stringbuilder(v=VS.100).aspx

内存分配部分。