.NetCore利用BlockingCollection实现简易消息队列_实用技巧_

 using (BlockingCollection bc = new BlockingCollection()) { Task.Factory.StartNew(() => { for (int i = 0; i < 1000; i++) { bc.Add(i); Thread.Sleep(50); } // Need to do this to keep foreach below from hanging bc.CompleteAdding(); }); // Now consume the blocking collection with foreach. // Use bc.GetConsumingEnumerable() instead of just bc because the // former will block waiting for completion and the latter will // simply take a snapshot of the current state of the underlying collection. foreach (var item in bc.GetConsumingEnumerable()) { Console.WriteLine(item); } } 

 /// 
 /// BlockingCollection演示消息队列 /// 
 ///  public class DemoQueueBlock where T : class { private static BlockingCollection Colls; public DemoQueueBlock() { } public static bool IsComleted() { if (Colls != null && Colls.IsCompleted) { return true; } return false; } public static bool HasEle() { if (Colls != null && Colls.Count>0) { return true; } return false; } public static bool Add(T msg) { if (Colls == null) { Colls = new BlockingCollection(); } Colls.Add(msg); return true; } public static T Take() { if (Colls == null) { Colls = new BlockingCollection(); } return Colls.Take(); } } /// 
 /// 消息体 /// 
 public class DemoMessage { public string BusinessType { get; set; } public string BusinessId { get; set; } public string Body { get; set; } } 

 //添加元素 while (true) { Console.WriteLine("请输入队列"); var read = Console.ReadLine(); if (read == "exit") { return; } DemoQueueBlock.Add(new DemoMessage() { BusinessId = read }); } 

 Task.Factory.StartNew(() => { //从队列中取元素。 while (!DemoQueueBlock.IsComleted()) { try { var m = DemoQueueBlock.Take(); Console.WriteLine("已消费:" + m.BusinessId); } catch (Exception ex) { Console.WriteLine(ex.Message); } } });