TPoller.cs 1001 B

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  1. using System;
  2. using System.Collections.Concurrent;
  3. using System.Collections.Generic;
  4. using Common.Base;
  5. using MongoDB.Bson;
  6. namespace TNet
  7. {
  8. public class TPoller : IPoller
  9. {
  10. // 线程同步队列,发送接收socket回调都放到该队列,由poll线程统一执行
  11. private readonly BlockingCollection<Action> blockingCollection = new BlockingCollection<Action>();
  12. public void Add(Action action)
  13. {
  14. this.blockingCollection.Add(action);
  15. }
  16. public void Dispose()
  17. {
  18. }
  19. public void Run(int timeout)
  20. {
  21. // 处理读写线程的回调
  22. Action action;
  23. if (this.blockingCollection.TryTake(out action, timeout))
  24. {
  25. var queue = new Queue<Action>();
  26. queue.Enqueue(action);
  27. while (true)
  28. {
  29. if (!this.blockingCollection.TryTake(out action, 0))
  30. {
  31. break;
  32. }
  33. queue.Enqueue(action);
  34. }
  35. while (queue.Count > 0)
  36. {
  37. Action a = queue.Dequeue();
  38. a();
  39. }
  40. }
  41. }
  42. }
  43. }