Queue.php 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
  1. <?php
  2. namespace utils;
  3. use traits\ErrorTrait;
  4. use think\facade\Config;
  5. use think\facade\Queue as QueueThink;
  6. use think\facade\Log;
  7. /**
  8. * Class Queue
  9. * @package crmeb\utils
  10. * @method $this do(string $do) 设置任务执行方法
  11. * @method $this job(string $job) 设置任务执行类名
  12. * @method $this errorCount(int $errorCount) 执行失败次数
  13. * @method $this data(...$data) 执行数据
  14. * @method $this secs(int $secs) 延迟执行秒数
  15. * @method $this log($log) 记录日志
  16. */
  17. class Queue
  18. {
  19. use ErrorTrait;
  20. /**
  21. * 任务执行
  22. * @var string
  23. */
  24. protected $do = 'doJob';
  25. /**
  26. * 默认任务执行方法名
  27. * @var string
  28. */
  29. protected $defaultDo;
  30. /**
  31. * 任务类名
  32. * @var string
  33. */
  34. protected $job;
  35. /**
  36. * 错误次数
  37. * @var int
  38. */
  39. protected $errorCount = 3;
  40. /**
  41. * 数据
  42. * @var array|string
  43. */
  44. protected $data;
  45. /**
  46. * 任务名
  47. * @var null
  48. */
  49. protected $queueName = null;
  50. /**
  51. * 延迟执行秒数
  52. * @var int
  53. */
  54. protected $secs = 0;
  55. /**
  56. * 记录日志
  57. * @var string|callable|array
  58. */
  59. protected $log;
  60. /**
  61. * @var array
  62. */
  63. protected $rules = ['do', 'data', 'errorCount', 'job', 'secs', 'log'];
  64. /**
  65. * @var static
  66. */
  67. protected static $instance;
  68. /**
  69. * Queue constructor.
  70. */
  71. protected function __construct()
  72. {
  73. $this->defaultDo = $this->do;
  74. }
  75. /**
  76. * @return static
  77. */
  78. public static function instance()
  79. {
  80. if (is_null(self::$instance)) {
  81. self::$instance = new static();
  82. }
  83. return self::$instance;
  84. }
  85. /**
  86. * 放入消息队列
  87. * @param array|null $data
  88. * @return mixed
  89. */
  90. public function push(?array $data = null)
  91. {
  92. if (!$this->job) {
  93. return $this->setError('需要执行的队列类必须存在');
  94. }
  95. $res = QueueThink::{$this->action()}(...$this->getValues($data));
  96. if(!$res){
  97. $res = QueueThink::{$this->action()}(...$this->getValues($data));
  98. if(!$res){
  99. Log::error('加入队列失败,参数:'.json_encode($this->getValues($data)));
  100. }
  101. }
  102. $this->clean();
  103. return $res;
  104. }
  105. /**
  106. * 清除数据
  107. */
  108. public function clean()
  109. {
  110. $this->secs = 0;
  111. $this->data = [];
  112. $this->log = null;
  113. $this->queueName = null;
  114. $this->errorCount = 3;
  115. $this->do = $this->defaultDo;
  116. }
  117. /**
  118. * 获取任务方式
  119. * @return string
  120. */
  121. protected function action()
  122. {
  123. return $this->secs ? 'later' : 'push';
  124. }
  125. /**
  126. * 获取参数
  127. * @param $data
  128. * @return array
  129. */
  130. protected function getValues($data)
  131. {
  132. $jobData['data'] = $data ?: $this->data;
  133. $jobData['do'] = $this->do;
  134. $jobData['errorCount'] = $this->errorCount;
  135. $jobData['log'] = $this->log;
  136. if ($this->do != $this->defaultDo) {
  137. $this->job .= '@' . Config::get('queue.prefix', 'eb_') . $this->do;
  138. }
  139. if ($this->secs) {
  140. return [$this->secs, $this->job, $jobData, $this->queueName];
  141. } else {
  142. return [$this->job, $jobData, $this->queueName];
  143. }
  144. }
  145. /**
  146. * @param $name
  147. * @param $arguments
  148. * @return $this
  149. */
  150. public function __call($name, $arguments)
  151. {
  152. if (in_array($name, $this->rules)) {
  153. if ($name === 'data') {
  154. $this->{$name} = $arguments;
  155. } else {
  156. $this->{$name} = $arguments[0] ?? null;
  157. }
  158. return $this;
  159. } else {
  160. throw new \RuntimeException('Method does not exist' . __CLASS__ . '->' . $name . '()');
  161. }
  162. }
  163. }