queue.php 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586
  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. Log::record("queue on looper......",Log::DEBUG);
  33. $content = $worker->pop($queues,1);
  34. if(is_array($content))
  35. {
  36. $method = key($content);
  37. $arg = current($content);
  38. $argx = json_encode($arg,JSON_UNESCAPED_UNICODE);
  39. Log::record("method={$method} args={$argx}",Log::DEBUG);
  40. $result = $logic_queue->$method($arg);
  41. if (!$result['state']) {
  42. $this->log($result['msg'],false);
  43. }
  44. $empty_times = 0;
  45. }
  46. else
  47. {
  48. $empty_times ++;
  49. if($empty_times > 600) {
  50. $model = Model();
  51. $model->checkActive();
  52. unset($model);
  53. $empty_times = 0;
  54. }
  55. }
  56. }
  57. }
  58. private function sig_handler($signo)
  59. {
  60. Log::record("queue quit at sig_handler.",Log::DEBUG);
  61. switch($signo) {
  62. case SIGINT:
  63. case SIGHUP:
  64. case SIGQUIT:
  65. case SIGTERM:
  66. $this->_stop = true;
  67. break;
  68. default:
  69. break;
  70. }
  71. }
  72. }