Streamwriter，StringBuilder和并行循环

Question

对不起，大量的代码，我不能解释与less.Basically我试图从许多任务写入文件。 你们能告诉我我做错了什么吗？ _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时，您还没有获得相同的锁定。

你的'修复'并不是一个真正的修复。你只能使问题更难以看清。它现在只会以不同的和更微妙的方式出错。例如，我假设您的for循环完成其最终迭代后，您的Write方法仍会继续调用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

看起来它不是并行性的问题。这是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

内存分配部分。

Answer 3

我认为原因是因为你在中并行支架访问StringBuilder的

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

和LogBuilder内执行锁（）将禁止在stringBuidler内存分配。你正在改变streamwriter来处理每个字符的日志，因此会让parellel进程解锁stringBuilder的内存分配。

·隔离并行处理成不同的行动通过TextWriter.Synchronized到可能会减少的问题

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 4

的代码可以更简单比@IanGriffiths回答在StreamWriter周围创建一个锁定包装。该框架在许多类上提供Synchronized helpers以在需要时创建线程安全实例。

我正在打印当前循环迭代变量i，因此可以很容易地看到并行性在调度中的影响。当作业开始时，必须使用局部变量来存储i的当前值。此技术将避免捕获闭包中的迭代变量的引用，每次打印变量，而不是所需的值1,2,3等。

有用的同步帮助程序的完整列表。 http://referencesource.microsoft.com/#q=Synchronized。

系统。类别

• 静态的ArrayList同步（ArrayList的列表）
• 静态IList的同步（IList的列表）
• 静态Hashtable的同步（哈希表表）
• 静态队列同步（队列队列）
• 静态排序列表同步（SortedList列表）
• 静态堆栈同步（堆栈堆栈）

System.Collections.Generic

• 静态IList的同步（名单列表）

System.IO

• 静流同步（流流）
• 静态TextReader同步（TextReader阅读器）
• 静态的TextWriter同步（TextWriter的作家）

System.Text.RegularExpressions

• 静态匹配同步（匹配度内）
• 静态组同步（集团内）

具有正确封装类的简单代码。

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++) 
     { 
      int local_i = i; 
      acts.Add(() => 
      { 
       var delay = rnd.Next(5); 
       Thread.Sleep(delay); 
       return local_i.ToString() + " act that that lasted " + delay.ToString(); 
      }); 
     } 
     try 
     { 
      Parallel.ForEach(acts, act => 
      { 
       _log.WriteLine(act.Invoke()); 
      }); 
     } 
     finally 
     { 
      _log.Dispose(); 
     } 
    } 
} 

public class LogBuilder : IDisposable 
{ 
    private FileStream _fileStream; 
    private StreamWriter _streamWriter; 
    private TextWriter _synchronizedWriter; 

    public LogBuilder() 
    { 
     _fileStream = new FileStream(@"C:\temp\log.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.ReadWrite); 
     _streamWriter = new StreamWriter(_fileStream) { AutoFlush = true }; 
     _synchronizedWriter = TextWriter.Synchronized(_streamWriter); 
    } 
    public void WriteLine(string message) 
    { 
     _synchronizedWriter.WriteLine(message); 
    } 

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