下面由laravel教程栏目给大家介绍laravel-swoole消息队列，希望对需要的朋友有所帮助！
这段时间用laravel8+laravel-swoole做项目,可发现laravel-swoole的扩展不兼容消息队列;
思来想去这咋办呢,这咋办呢.咋办那就自己写咯!还好thinkphp-swoole扩展已经兼容了,那不就嘿嘿嘿!
直接上修改的思路和代码!开干!
一种是增加另外启动的命令或者在swoole启动的时候一起启动消息队列进行消费,我这么懒的人一个命令能解决的,绝不写两命令.
首先重写swoole启动命令
isRunning()) { $this->error('Failed! swoole_http_server process is already running.'); return; } $host = Arr::get($this->config, 'server.host'); $port = Arr::get($this->config, 'server.port'); $hotReloadEnabled = Arr::get($this->config, 'hot_reload.enabled'); $queueEnabled = Arr::get($this->config, 'queue.enabled'); $accessLogEnabled = Arr::get($this->config, 'server.access_log'); $coroutineEnable = Arr::get($this->config, 'coroutine.enable'); $this->info('Starting swoole http server...'); $this->info("Swoole http server started: "); if ($this->isDaemon()) { $this->info( '> (You can run this command to ensure the ' . 'swoole_http_server process is running: ps aux|grep "swoole")' ); } $manager = $this->laravel->make(Manager::class); $server = $this->laravel->make(Server::class); if ($accessLogEnabled) { $this->registerAccessLog(); } //热更新重写 if ($hotReloadEnabled) { $manager->addProcess($this->getHotReloadProcessNow($server)); } //启动消息队列进行消费 if ($queueEnabled) { $this->prepareQueue($manager); } if ($coroutineEnable) { Runtime::enableCoroutine(true, Arr::get($this->config, 'coroutine.flags', SWOOLE_HOOK_ALL)); } $manager->run(); } /** * @param Server $server * @return Process|void */ protected function getHotReloadProcessNow($server) { return new Process(function () use ($server) { $watcher = new FileWatcher( Arr::get($this->config, 'hot_reload.include', []), Arr::get($this->config, 'hot_reload.exclude', []), Arr::get($this->config, 'hot_reload.name', []) ); $watcher->watch(function () use ($server) { $server->reload(); }); }, false, 0, true); }}
InteractsWithQueue 类
laravel->make(QueueManager::class); $queueManager->attachToServer($manager, $this->output); }}
Manager类
轻舟办公
基于AI的智能办公平台
194 查看详情 
container = $container; } /** * @param ServerManager $server */ public function attachToServer(ServerManager $server, OutputStyle $output) { $this->output = $output; $this->listenForEvents(); $this->createWorkers(); foreach ($this->workers as $worker) { $server->addProcess(new Process($worker, false, 0, true)); } } /** * 运行消息队列命令 */ public function run(): void { @cli_set_process_title("swoole queue: manager process"); $this->listenForEvents(); $this->createWorkers(); $pool = new Pool(count($this->workers)); $pool->on(Constant::EVENT_WORKER_START, function (Pool $pool, int $workerId) { $process = $pool->getProcess($workerId); run($this->workers[$workerId], $process); }); $pool->start(); } /** * 创建执行任务 */ protected function createWorkers() { $workers = $this->getConfig('queue.workers', []); foreach ($workers as $queue => $options) { if (strpos($queue, '@') !== false) { [$queue, $connection] = explode('@', $queue); } else { $connection = null; } $this->workers[] = function (Process $process) use ($options, $connection, $queue) { @cli_set_process_title("swoole queue: worker process"); /** @var Worker $worker */ $worker = $this->container->make('queue.worker'); /** @var WorkerOptions $option */ $option = $this->container->make(WorkerOptions::class); $option->sleep = Arr::get($options, "sleep", 3); $option->maxTries = Arr::get($options, "tries", 0); $option->timeout = Arr::get($options, "timeout", 60); $timer = Timer::after($option->timeout * 1000, function () use ($process) { $process->exit(); }); $worker->runNextJob($connection, $queue, $option); Timer::clear($timer); }; } } /** * 注册事件 */ protected function listenForEvents() { $this->container->make('events')->listen(JobFailed::class, function (JobFailed $event) { $this->writeOutput($event->job); $this->logFailedJob($event); }); } /** * 记录失败任务 * @param JobFailed $event */ protected function logFailedJob(JobFailed $event) { $this->container['queue.failer']->log( $event->connection, $event->job->getQueue(), $event->job->getRawBody(), $event->exception ); } /** * Write the status output for the queue worker. * * @param Job $job * @param $status */ protected function writeOutput(Job $job, $status) { switch ($status) { case 'starting': $this->writeStatus($job, 'Processing', 'comment'); break; case 'success': $this->writeStatus($job, 'Processed', 'info'); break; case 'failed': $this->writeStatus($job, 'Failed', 'error'); break; } } /** * Format the status output for the queue worker. * * @param Job $job * @param string $status * @param string $type * @return void */ protected function writeStatus(Job $job, $status, $type) { $this->output->writeln(sprintf( "[%s][%s] %s %s", date('Y-m-d H:i:s'), $job->getJobId(), str_pad("{$status}:", 11), $job->getName() )); }}
增加CrmebServiceProvider类
app->singleton(Manager::class, function ($app) { return new Manager($app, 'laravel'); }); $this->app->alias(Manager::class, 'swoole.manager'); $this->app->singleton('queue.worker', function ($app) { $isDownForMaintenance = function () { return $this->app->isDownForMaintenance(); }; return new Worker( $app['queue'], $app['events'], $app[ExceptionHandler::class], $isDownForMaintenance ); }); } /** * Boot websocket routes. * * @return void */ protected function bootWebsocketRoutes() { require base_path('vendor/swooletw/laravel-swoole') . '/routes/laravel_routes.php'; } /** * Register access log middleware to container. * * @return void */ protected function pushAccessLogMiddleware() { $this->app->make(Kernel::class)->pushMiddleware(AccessLog::class); } /** * Register commands. */ protected function registerCommands() { $this->commands([ HttpServerCommand::class, ]); } /** * Merge configurations. */ protected function mergeConfigs() { $this->mergeConfigFrom(base_path('vendor/swooletw/laravel-swoole') . '/config/swoole_http.php', 'swoole_http'); $this->mergeConfigFrom(base_path('vendor/swooletw/laravel-swoole') . '/config/swoole_websocket.php', 'swoole_websocket'); } /** * Publish files of this package. */ protected function publishFiles() { $this->publishes([ base_path('vendor/swooletw/laravel-swoole') . '/config/swoole_http.php' => base_path('config/swoole_http.php'), base_path('vendor/swooletw/laravel-swoole') . '/config/swoole_websocket.php' => base_path('config/swoole_websocket.php'), base_path('vendor/swooletw/laravel-swoole') . '/routes/websocket.php' => base_path('routes/websocket.php'), ], 'laravel-swoole'); }}
然后再把crmebswooleCrmebServiceProvider::class放入config/app.php中的providers中加载重写了swoole的命令启动方式
配置config/swoole_http.php
return [ 'queue' => [ //是否开启自动消费队列 'enabled' => true, 'workers' => [ //队列名称 'CRMEB' => [] ] ],];
输入命令:php artisan crmeb:http restart
swoole启动后就可以自动消费队列了。
相关推荐：最新的五个Laravel视频教程
以上就是laravel8中laravel-swoole的扩展不兼容消息队列怎么办？的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/627905.html
微信扫一扫 
支付宝扫一扫 
