Queue.php 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  1. <?php
  2. // +----------------------------------------------------------------------
  3. // | CRMEB [ CRMEB赋能开发者,助力企业发展 ]
  4. // +----------------------------------------------------------------------
  5. // | Copyright (c) 2016~2023 https://www.crmeb.com All rights reserved.
  6. // +----------------------------------------------------------------------
  7. // | Licensed CRMEB并不是自由软件,未经许可不能去掉CRMEB相关版权
  8. // +----------------------------------------------------------------------
  9. // | Author: CRMEB Team <admin@crmeb.com>
  10. // +----------------------------------------------------------------------
  11. namespace crmeb\utils;
  12. use think\facade\Config;
  13. use think\facade\Queue as QueueThink;
  14. use think\facade\Log;
  15. /**
  16. * Class Queue
  17. * @package crmeb\utils
  18. * @method $this do(string $do) 设置任务执行方法
  19. * @method $this job(string $job) 设置任务执行类名
  20. * @method $this errorCount(int $errorCount) 执行失败次数
  21. * @method $this data(...$data) 执行数据
  22. * @method $this secs(int $secs) 延迟执行秒数
  23. * @method $this log($log) 记录日志
  24. */
  25. class Queue
  26. {
  27. /**
  28. * 错误信息
  29. * @var string
  30. */
  31. protected $error;
  32. /**
  33. * 设置错误信息
  34. * @param string|null $error
  35. * @return bool
  36. */
  37. protected function setError(?string $error = null)
  38. {
  39. $this->error = $error ?: '未知错误';
  40. return false;
  41. }
  42. /**
  43. * 获取错误信息
  44. * @return string
  45. */
  46. public function getError()
  47. {
  48. $error = $this->error;
  49. $this->error = null;
  50. return $error;
  51. }
  52. /**
  53. * 任务执行
  54. * @var string
  55. */
  56. protected $do = 'doJob';
  57. /**
  58. * 默认任务执行方法名
  59. * @var string
  60. */
  61. protected $defaultDo;
  62. /**
  63. * 任务类名
  64. * @var string
  65. */
  66. protected $job;
  67. /**
  68. * 错误次数
  69. * @var int
  70. */
  71. protected $errorCount = 3;
  72. /**
  73. * 数据
  74. * @var array|string
  75. */
  76. protected $data;
  77. /**
  78. * 队列名
  79. * @var null
  80. */
  81. protected $queueName = null;
  82. /**
  83. * 延迟执行秒数
  84. * @var int
  85. */
  86. protected $secs = 0;
  87. /**
  88. * 记录日志
  89. * @var string|callable|array
  90. */
  91. protected $log;
  92. /**
  93. * @var array
  94. */
  95. protected $rules = ['do', 'data', 'errorCount', 'job', 'secs', 'log'];
  96. /**
  97. * @var static
  98. */
  99. protected static $instance;
  100. /**
  101. * Queue constructor.
  102. */
  103. protected function __construct()
  104. {
  105. $this->defaultDo = $this->do;
  106. }
  107. /**
  108. * @return static
  109. */
  110. public static function instance()
  111. {
  112. if (is_null(self::$instance)) {
  113. self::$instance = new static();
  114. }
  115. return self::$instance;
  116. }
  117. /**
  118. * 设置列名
  119. * @param string $queueName
  120. * @return $this
  121. */
  122. public function setQueueName(string $queueName)
  123. {
  124. $this->queueName = $queueName;
  125. return $this;
  126. }
  127. /**
  128. * 放入消息队列
  129. * @param array|null $data
  130. * @return mixed
  131. */
  132. public function push(?array $data = null)
  133. {
  134. if (!$this->job) {
  135. return $this->setError('需要执行的队列类必须存在');
  136. }
  137. $jodValue = $this->getValues($data);
  138. $res = QueueThink::{$this->action()}(...$jodValue);
  139. if (!$res) {
  140. $res = QueueThink::{$this->action()}(...$jodValue);
  141. if (!$res) {
  142. Log::error('加入队列失败,参数:' . json_encode($this->getValues($data)));
  143. }
  144. }
  145. $this->clean();
  146. return $res;
  147. }
  148. /**
  149. * 清除数据
  150. */
  151. public function clean()
  152. {
  153. $this->secs = 0;
  154. $this->data = [];
  155. $this->log = null;
  156. $this->queueName = null;
  157. $this->errorCount = 3;
  158. $this->do = $this->defaultDo;
  159. }
  160. /**
  161. * 获取任务方式
  162. * @return string
  163. */
  164. protected function action()
  165. {
  166. return $this->secs ? 'later' : 'push';
  167. }
  168. /**
  169. * 获取参数
  170. * @param $data
  171. * @return array
  172. */
  173. protected function getValues($data)
  174. {
  175. $jobData['data'] = $data ?: $this->data;
  176. $jobData['do'] = $this->do;
  177. $jobData['errorCount'] = $this->errorCount;
  178. $jobData['log'] = $this->log;
  179. if ($this->do != $this->defaultDo) {
  180. $this->job .= '@' . Config::get('queue.prefix', 'eb_') . $this->do;
  181. }
  182. if ($this->secs) {
  183. return [$this->secs, $this->job, $jobData, $this->queueName];
  184. } else {
  185. return [$this->job, $jobData, $this->queueName];
  186. }
  187. }
  188. /**
  189. * @param $name
  190. * @param $arguments
  191. * @return $this
  192. */
  193. public function __call($name, $arguments)
  194. {
  195. if (in_array($name, $this->rules)) {
  196. if ($name === 'data') {
  197. $this->{$name} = $arguments;
  198. } else {
  199. $this->{$name} = $arguments[0] ?? null;
  200. }
  201. return $this;
  202. } else {
  203. throw new \RuntimeException('Method does not exist' . __CLASS__ . '->' . $name . '()');
  204. }
  205. }
  206. }