c# 多线程排队队列实现的源码

时间:2023-03-09 06:48:47
c# 多线程排队队列实现的源码
[csharp] view plaincopy
  1. using System;
  2. using System.Threading;
  3. using System.Collections;
  4. using System.Collections.Generic;
  5. // 将线程同步事件封装在此类中,
  6. // 以便于将这些事件传递给 Consumer 和
  7. // Producer 类。
  8. public class SyncEvents
  9. {
  10. public SyncEvents()
  11. {
  12. // AutoResetEvent 用于“新项”事件,因为
  13. // 我们希望每当使用者线程响应此事件时,
  14. // 此事件就会自动重置。
  15. _newItemEvent = new AutoResetEvent(false);
  16. // ManualResetEvent 用于“退出”事件,因为
  17. // 我们希望发出此事件的信号时有多个线程响应。
  18. // 如果使用 AutoResetEvent,事件
  19. // 对象将在单个线程作出响应之后恢复为
  20. // 未发信号的状态,而其他线程将
  21. // 无法终止。
  22. _exitThreadEvent = new ManualResetEvent(false);
  23. // 这两个事件也放在一个 WaitHandle 数组中,以便
  24. // 使用者线程可以使用 WaitAny 方法
  25. // 阻塞这两个事件。
  26. _eventArray = new WaitHandle[2];
  27. _eventArray[0] = _newItemEvent;
  28. _eventArray[1] = _exitThreadEvent;
  29. }
  30. // 公共属性允许对事件进行安全访问。
  31. public EventWaitHandle ExitThreadEvent
  32. {
  33. get { return _exitThreadEvent; }
  34. }
  35. public EventWaitHandle NewItemEvent
  36. {
  37. get { return _newItemEvent; }
  38. }
  39. public WaitHandle[] EventArray
  40. {
  41. get { return _eventArray; }
  42. }
  43. private EventWaitHandle _newItemEvent;
  44. private EventWaitHandle _exitThreadEvent;
  45. private WaitHandle[] _eventArray;
  46. }
  47. // Producer 类(使用一个辅助线程)
  48. // 将项异步添加到队列中,共添加 20 个项。
  49. public class Producer
  50. {
  51. public Producer(Queue<int> q, SyncEvents e)
  52. {
  53. _queue = q;
  54. _syncEvents = e;
  55. }
  56. public void ThreadRun()
  57. {
  58. int count = 0;
  59. Random r = new Random();
  60. while (!_syncEvents.ExitThreadEvent.WaitOne(0, false))
  61. {
  62. lock (((ICollection)_queue).SyncRoot)
  63. {
  64. while (_queue.Count < 20)
  65. {
  66. _queue.Enqueue(r.Next(0, 100));
  67. _syncEvents.NewItemEvent.Set();
  68. count++;
  69. }
  70. }
  71. }
  72. Console.WriteLine("Producer thread: produced {0} items", count);
  73. }
  74. private Queue<int> _queue;
  75. private SyncEvents _syncEvents;
  76. }
  77. // Consumer 类通过自己的辅助线程使用队列
  78. // 中的项。Producer 类使用 NewItemEvent
  79. // 将新项通知 Consumer 类。
  80. public class Consumer
  81. {
  82. public Consumer(Queue<int> q, SyncEvents e)
  83. {
  84. _queue = q;
  85. _syncEvents = e;
  86. }
  87. public void ThreadRun()
  88. {
  89. int count = 0;
  90. while (WaitHandle.WaitAny(_syncEvents.EventArray) != 1)
  91. {
  92. lock (((ICollection)_queue).SyncRoot)
  93. {
  94. int item = _queue.Dequeue();
  95. }
  96. count++;
  97. }
  98. Console.WriteLine("Consumer Thread: consumed {0} items", count);
  99. }
  100. private Queue<int> _queue;
  101. private SyncEvents _syncEvents;
  102. }
  103. public class ThreadSyncSample
  104. {
  105. private static void ShowQueueContents(Queue<int> q)
  106. {
  107. // 对集合进行枚举本来就不是线程安全的,
  108. // 因此在整个枚举过程中锁定集合以防止
  109. // 使用者和制造者线程修改内容
  110. // 是绝对必要的。(此方法仅由
  111. // 主线程调用。)
  112. lock (((ICollection)q).SyncRoot)
  113. {
  114. foreach (int i in q)
  115. {
  116. Console.Write("{0} ", i);
  117. }
  118. }
  119. Console.WriteLine();
  120. }
  121. static void Main()
  122. {
  123. // 配置结构,该结构包含线程同步
  124. // 所需的事件信息。
  125. SyncEvents syncEvents = new SyncEvents();
  126. // 泛型队列集合用于存储要制造和使用的
  127. // 项。此例中使用的是“int”。
  128. Queue<int> queue = new Queue<int>();
  129. // 创建对象,一个用于制造项,一个用于
  130. // 使用项。将队列和线程同步事件传递给
  131. // 这两个对象。
  132. Console.WriteLine("Configuring worker threads...");
  133. Producer producer = new Producer(queue, syncEvents);
  134. Consumer consumer = new Consumer(queue, syncEvents);
  135. // 为制造者对象和使用者对象创建线程
  136. // 对象。此步骤并不创建或启动
  137. // 实际线程。
  138. Thread producerThread = new Thread(producer.ThreadRun);
  139. Thread consumerThread = new Thread(consumer.ThreadRun);
  140. // 创建和启动两个线程。
  141. Console.WriteLine("Launching producer and consumer threads...");
  142. producerThread.Start();
  143. consumerThread.Start();
  144. // 为制造者线程和使用者线程设置 10 秒的运行时间。
  145. // 使用主线程(执行此方法的线程)
  146. // 每隔 2.5 秒显示一次队列内容。
  147. for (int i = 0; i < 4; i++)
  148. {
  149. Thread.Sleep(2500);
  150. ShowQueueContents(queue);
  151. }
  152. // 向使用者线程和制造者线程发出终止信号。
  153. // 这两个线程都会响应,由于 ExitThreadEvent 是
  154. // 手动重置的事件,因此除非显式重置,否则将保持“设置”。
  155. Console.WriteLine("Signaling threads to terminate...");
  156. syncEvents.ExitThreadEvent.Set();
  157. // 使用 Join 阻塞主线程,首先阻塞到制造者线程
  158. // 终止,然后阻塞到使用者线程终止。
  159. Console.WriteLine("main thread waiting for threads to finish...");
  160. producerThread.Join();
  161. consumerThread.Join();
  162. }
  163. }
  1. namespace WindowsFormsApplication1
  2. {
  3. public partial class Form3 : Form
  4. {
  5. public Form3()
  6. {
  7. InitializeComponent();
  8. }
  9. public delegate void Delegate1();
  10. public delegate void Delegate2(DataTable dt);
  11. public void buttonFind_Click(object sender, EventArgs e)
  12. {
  13. Delegate1 d1 = new Delegate1(Find);
  14. d1.BeginInvoke(new AsyncCallback(AsyncCallback1), d1);
  15. }
  16. public void AsyncCallback1(IAsyncResult iAsyncResult)
  17. {
  18. Delegate1 d1 = (Delegate1)iAsyncResult.AsyncState;
  19. d1.EndInvoke(iAsyncResult);
  20. }
  21. public void Find()
  22. {
  23. DataTable dt = new DataTable();
  24. dt.Columns.Add("name", typeof(string));
  25. dt.Columns.Add("age", typeof(int));
  26. AddRow(dt, "张三", 19);
  27. AddRow(dt, "张三", 19);
  28. AddRow(dt, "李四", 18);
  29. this.Invoke(new Delegate2(Bind2), new object[] { dt });
  30. }
  31. public void AddRow(DataTable dt, string name, int age)
  32. {
  33. DataRow dr = dt.NewRow();
  34. dr["name"] = name;
  35. dr["age"] = age;
  36. dt.Rows.Add(dr);
  37. }
  38. public void Bind2(DataTable dt)
  39. {
  40. this.dataGridView1.DataSource = dt;
  41. }
  42. }

http://blog.****.net/loveandangle/article/details/6733642