C# 生产者与消费者模式
阅读原文时间:2023年07月13日阅读:1

情景:一个线程不断获取数据,另一个线程不断处理这些数据。

常规方法:数据列表加锁,两个线程获取锁,拿到操作权;类似代码如下:(不推荐)

static void Main(string[] args)
{
lockClass l = new lockClass();
for (int i = 0; i < 1000000; i++)
{
l.Equeue(i.ToString());

        }  
   }

public class lockClass
{
Queue currentQueue = new Queue(10000000);//当前要插入数据的队列
static readonly object objlock = new object();
FileStream f = new FileStream("D://1.txt", FileMode.Create, FileAccess.Write, FileShare.None);
StreamWriter writer;
public lockClass()
{
writer = new StreamWriter(f);
var backgroundWorker = new BackgroundWorker();
backgroundWorker.DoWork += backgroundWorker_DoWork;
backgroundWorker.RunWorkerAsync();
}
void backgroundWorker_DoWork(object sender, DoWorkEventArgs e)
{
while (true)
{
lock (objlock)
{
if (currentQueue.Count > 0)
{
var item = currentQueue.Dequeue();
Console.WriteLine(item);
writer.WriteLine(item);

                }

            }

        }  
    }

    public void Equeue(string item)  
    {  
        lock (objlock)  
        {

            currentQueue.Enqueue(item);

        }  
    }

}

  方法2:双缓存队列处理,意思就是说,用两个队列,一个队列用于获取数据,另一个队列用于操作数据,通过信号量来处理线程调度,来取消“锁”带来的资源切换浪费,参考代码如下:

static void Main(string[] args)
{
var test = new DoubleBufferedQueue();
for (int i = 0; i < 1000000; i++)
{
test.Equeue(i.ToString());

        }  

}

public class DoubleBufferedQueue
{
public readonly Queue Queue1 = new Queue(10000000);
public readonly Queue Queue2 = new Queue(10000000);
private readonly ManualResetEvent lock1 = new ManualResetEvent(true);//一开始可以执行
private readonly ManualResetEvent lock2 = new ManualResetEvent(false);
private readonly AutoResetEvent _autoReset = new AutoResetEvent(true);
private volatile Queue currentQueue = new Queue(10000000);//当前要插入数据的队列
FileStream f = new FileStream("D://1.txt", FileMode.Create, FileAccess.Write, FileShare.None);
StreamWriter writer;
public DoubleBufferedQueue()
{
writer = new StreamWriter(f);
currentQueue = Queue1;
var backgroundWorker = new BackgroundWorker();
backgroundWorker.DoWork += backgroundWorker_DoWork;
backgroundWorker.RunWorkerAsync();
}

   void backgroundWorker\_DoWork(object sender, DoWorkEventArgs e)  
   {  
       while (true)  
       {  
           this.\_autoReset.WaitOne();//没有成员入队列时不进行其他操作;  
           this.lock2.Reset();  
           this.lock1.WaitOne();  
           var readQueue = currentQueue;  
           currentQueue = (currentQueue == Queue1) ? Queue2 : Queue1;  
           this.lock2.Set();  
           writeToConsonle(currentQueue);

       }  
   }

   void writeToConsonle(Queue<string> readQueue)  
   {

       while (readQueue.Count > 0)  
       {  
          var item= readQueue.Dequeue();  
          Console.WriteLine(item);  
          writer.WriteLine(item);  
       }  
   }

   public void Equeue(string item)  
   {  
       this.lock2.WaitOne();  
       this.lock1.Reset();  
       currentQueue.Enqueue(item);  
       lock1.Set();  
       \_autoReset.Set();  
   }

}

  方法3:用微软提供的BlockingCollection(线程安全的,可阻塞的资源的),个人理解就是资源安全的队列,并且当没有操作的时候(队列空闲的时候)不耗费资源,个人觉得和方法2原理类似(推荐使用)

static void Main(string[] args)
{

var block = new blockingCollectionClass();
for (int i = 0; i < 10000; i++)
{

            block.Add(i.ToString());

        }  
       Console.ReadKey();  

}

public class blockingCollectionClass
{
BlockingCollection blockingCollection = new BlockingCollection();
FileStream f = new FileStream("D://1.txt", FileMode.Create, FileAccess.Write, FileShare.None);
StreamWriter writer;
public void Add(string Item)
{
blockingCollection.Add(Item);
}

    public blockingCollectionClass()  
    {  
        writer = new StreamWriter(f);  
        var backgroundWorker = new BackgroundWorker();  
        backgroundWorker.DoWork += backgroundWorker\_DoWork;  
        backgroundWorker.RunWorkerAsync();  
    }

    void backgroundWorker\_DoWork(object sender, DoWorkEventArgs e)  
    {  
        foreach (string value in blockingCollection.GetConsumingEnumerable())  
        {  
            Console.WriteLine(value);  
            writer.WriteLine(value);  
        }  
    }

}

  情景2:秒杀活动、抢票等活动时,并发性很高,导致服务器阻塞,用户请求丢失;

策略1:可以采用以上队列的形式处理服务器高并发问题,所有的请求先加入队列,排队,后台线程来处理队列里面的请求;

策略2:够建一个队列容器,接收请求的线程从容器中取一个空的对列,当队列填满后,放回到容器中,再次从容器中取一个空队列;处理线程需要从容器中取出非空的队列,处理完队列为空,放回到容器去;从容器中取放队列需要加锁。如果要保证处理的顺序,容器可以选队列(放队列的队列);