c# - .NET 的线程安全缓冲区

Question

(注意:虽然我想要关于 .Net 4.0 future 的想法，但对于这个项目我仅限于 .Net 3.5。)

我有一个线程，它正在从外部设备异步读取数据(在代码示例中由非常有创意的 strSomeData 模拟 :-) 并将其存储在 StringBuilder '缓冲区' (strBuilderBuffer :-)

在“主要代码”中，我想“蚕食”这个“缓冲区”。但是，从“操作”的角度来看，我不确定如何以线程安全的方式执行此操作。我理解从“数据”的角度来看它是安全的，因为根据 msdn，“此 (StringBuilder) 类型的任何公共(public)静态成员都是线程安全的。不保证任何实例成员都是线程安全的。 “但是，我下面的代码表明，从“操作”的角度来看，它可能不是线程安全的。

关键是我担心两行代码:

string strCurrentBuffer = ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer.ToString();
// Thread 'randomly' slept due to 'inconvenient' comp resource scheduling...
ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer.Length = 0;

如果计算机操作系统在“读取”缓冲区和“清除”缓冲区之间让我的线程休眠，我可能会丢失数据(这很糟糕 :-(

有什么方法可以保证“原子性”吗？这两行并强制计算机不打断它们？

关于下面 Vlad 关于使用 lock 的建议，我试过了，但没有用(完全没有用):

    public void BufferAnalyze()
    {
        String strCurrentBuffer;
        lock (ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer)
        {
            strCurrentBuffer = ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer.ToString();
            Console.WriteLine("[BufferAnalyze()]  ||<<  Thread 'Randomly' Slept due to comp resource scheduling");
            Thread.Sleep(1000);  //  Simulate poor timing of thread resourcing...
            ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer.Length = 0;
        }
        Console.WriteLine("[BufferAnalyze()]\r\nstrCurrentBuffer[{0}] == {1}", strCurrentBuffer.Length.ToString(), strCurrentBuffer);
    }

有没有更好的方法来实现线程安全缓冲区？

完整代码如下:

namespace ExploringThreads
{
    /// <summary>
    /// Description of BasicThreads_TestThreadSafety_v1a
    /// </summary>
    class ThreadWorker_TestThreadSafety_v1a
    {
        private Thread thread;
        public static StringBuilder strBuilderBuffer = new StringBuilder("", 7500);
        public static StringBuilder strBuilderLog = new StringBuilder("", 7500);

        public bool IsAlive
        {
            get { return thread.IsAlive; }
        }

        public ThreadWorker_TestThreadSafety_v1a(string strThreadName)
        {
            // It is possible to have a thread begin execution as soon as it is created.
            // In the case of MyThread this is done by instantiating a Thread object inside MyThread's constructor.
            thread = new Thread(new ThreadStart(this.threadRunMethod));
            thread.Name = strThreadName;
            thread.Start();
        }

        public ThreadWorker_TestThreadSafety_v1a() : this("")
        {
            //   NOTE:  constructor overloading ^|^
        }

        //Entry point of thread.
        public void threadRunMethod()
        {
            Console.WriteLine("[ThreadWorker_TestThreadSafety_v1a threadRunMethod()]");
            Console.WriteLine(thread.Name + " starting.");
            int intSomeCounter = 0;
            string strSomeData = "";
            do
            {
                Console.WriteLine("[ThreadWorker_TestThreadSafety_v1a threadRunMethod()] running.");
                intSomeCounter++;
                strSomeData = "abcdef" + intSomeCounter.ToString() + "|||";
                strBuilderBuffer.Append(strSomeData);
                strBuilderLog.Append(strSomeData);
                Thread.Sleep(200);
            } while(intSomeCounter < 15);

            Console.WriteLine(thread.Name + " terminating.");
        }
    }
    /// <summary>
    /// Description of BasicThreads_TestThreadSafety_v1a.
    /// </summary>
    public class BasicThreads_TestThreadSafety_v1a
    {
        public BasicThreads_TestThreadSafety_v1a()
        {
        }

        public void BufferAnalyze()
        {
            string strCurrentBuffer = ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer.ToString();
            Console.WriteLine("[BufferAnalyze()]  ||<<  Thread 'Randomly' Slept due to comp resource scheduling");
            Thread.Sleep(1000);  //  Simulate poor timing of thread resourcing...
            ThreadWorker_TestThreadSafety_v1a.strBuilderBuffer.Length = 0;
            Console.WriteLine("[BufferAnalyze()]\r\nstrCurrentBuffer[{0}] == {1}", strCurrentBuffer.Length.ToString(), strCurrentBuffer);
        }

        public void TestBasicThreads_TestThreadSafety_v1a()
        {
            Console.Write("Starting TestBasicThreads_TestThreadSafety_v1a  >>>  Press any key to continue . . . ");
            Console.Read();

            // First, construct a MyThread object.
            ThreadWorker_TestThreadSafety_v1a threadWorker_TestThreadSafety_v1a = new ThreadWorker_TestThreadSafety_v1a("threadWorker_TestThreadSafety_v1a Child");

            do
            {
                Console.WriteLine("[TestBasicThreads_TestThreadSafety_v1a()]");
                Thread.Sleep(750);
                BufferAnalyze();
                //} while (ThreadWorker_TestThreadSafety_v1a.thread.IsAlive);
            } while (threadWorker_TestThreadSafety_v1a.IsAlive);
            BufferAnalyze();
            Thread.Sleep(1250);
            Console.WriteLine("[TestBasicThreads_TestThreadSafety_v1a()]");
            Console.WriteLine("ThreadWorker_TestThreadSafety_v1a.strBuilderLog[{0}] == {1}", ThreadWorker_TestThreadSafety_v1a.strBuilderLog.Length.ToString(), ThreadWorker_TestThreadSafety_v1a.strBuilderLog);

            Console.Write("Completed TestBasicThreads_TestThreadSafety_v1a  >>>  Press any key to continue . . . ");
            Console.Read();
        }
    }
}

Answer 1

下载Reactive Extensions 3.5 的反向移植 here .还有一个NuGet包装它。下载后，只需在项目中引用 System.Threading.dll。

现在您也可以在 .NET 3.5 中使用 .NET 4.0 中的所有新并发集合标准。最适合您情况的是 BlockingCollection .它基本上是一个缓冲区，允许线程像普通队列一样将项目入队和出队。除了出队操作阻塞，直到一个项目可用。

无需使用 StringBuilder现在根本不上课。以下是我将如何重构您的代码。我尽量使示例简短，以便更容易理解。

public class Example
{
  private BlockingCollection<string> buffer = new BlockingCollection<string>();

  public Example()
  {
    new Thread(ReadFromExternalDevice).Start();
    new Thread(BufferAnalyze).Start();
  }

  private void ReadFromExteneralDevice()
  {
    while (true)
    {
      string data = GetFromExternalDevice();
      buffer.Add(data);
      Thread.Sleep(200);
    }
  }

  private void BufferAnalyze()
  {
    while (true)
    {
      string data = buffer.Take(); // This blocks if nothing is in the queue.
      Console.WriteLine(data);
    }
  } 
}

为了将来引用 BufferBlock<T>来自 TPL Data Flow 的类(class)library 将做与 BlockingCollection 基本相同的事情.它将在 .NET 4.5 中可用。