queue.php 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. <?php
  2. /**
  3. * 队列
  4. *
  5. *
  6. *
  7. *
  8. */
  9. defined('InShopNC') or exit('Access Invalid!');
  10. //此行代码会导致bug
  11. //ini_set('default_socket_timeout', -1);
  12. class queueControl extends BaseCronControl
  13. {
  14. private $_stop = false;
  15. public function indexOp()
  16. {
  17. if (ob_get_level()) ob_end_clean();
  18. pcntl_signal(SIGINT, array($this,'sig_handler'));
  19. pcntl_signal(SIGHUP, array($this,'sig_handler'));
  20. pcntl_signal(SIGQUIT, array($this,'sig_handler'));
  21. pcntl_signal(SIGTERM, array($this,'sig_handler'));
  22. $logic_queue = Logic('queue');
  23. $worker = new QueueServer();
  24. $queues = $worker->scan();
  25. $empty_times = 0;
  26. while (true)
  27. {
  28. pcntl_signal_dispatch();
  29. if ($this->_stop) {
  30. exit;
  31. }
  32. $content = $worker->pop($queues,1);
  33. if(is_array($content))
  34. {
  35. $method = key($content);
  36. $arg = current($content);
  37. $argx = json_encode($arg,JSON_UNESCAPED_UNICODE);
  38. Log::record("method={$method} args={$argx}",Log::DEBUG);
  39. $result = $logic_queue->$method($arg);
  40. if (!$result['state']) {
  41. $this->log($result['msg'],false);
  42. }
  43. $empty_times = 0;
  44. }
  45. else
  46. {
  47. $empty_times ++;
  48. if($empty_times > 600) {
  49. $model = Model();
  50. $model->checkActive();
  51. unset($model);
  52. $empty_times = 0;
  53. }
  54. }
  55. }
  56. }
  57. private function sig_handler($signo)
  58. {
  59. Log::record("queue quit at sig_handler.",Log::DEBUG);
  60. switch($signo) {
  61. case SIGINT:
  62. case SIGHUP:
  63. case SIGQUIT:
  64. case SIGTERM:
  65. $this->_stop = true;
  66. break;
  67. default:
  68. break;
  69. }
  70. }
  71. }