共计 8212 个字符，预计需要花费 21 分钟才能阅读完成。
这段时间用 laravel8+laravel-swoole 做我的项目, 可发现 laravel-swoole 的扩大不兼容音讯队列;
思来想去这咋办呢, 这咋办呢. 咋办那就本人写咯! 还好 thinkphp-swoole 扩大曾经兼容了, 那不就嘿嘿嘿!
间接上批改的思路和代码! 开干!
一种是减少另外启动的命令或者在 swoole 启动的时候一起启动音讯队列进行生产, 我这么懒的人一个命令能解决的, 绝不写两命令.
首先重写 swoole 启动命令
<?php
namespace crmeb\swoole\command;
use Illuminate\Support\Arr;
use Swoole\Process;
use SwooleTW\Http\Server\Facades\Server;
use SwooleTW\Http\Server\Manager;
use crmeb\swoole\server\InteractsWithQueue;
use crmeb\swoole\server\FileWatcher;
use Swoole\Runtime;
class HttpServerCommand extends \SwooleTW\Http\Commands\HttpServerCommand
{
use InteractsWithQueue;
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'crmeb:http {action : start|stop|restart|reload|infos}';
/**
* Run swoole_http_server.
*/
protected function start()
{if ($this->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: <http://{$host}:{$port}>");
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 类
<?php
namespace crmeb\swoole\server;
use crmeb\swoole\queue\Manager as QueueManager;
use SwooleTW\Http\Server\Manager;
/**
* Trait InteractsWithQueue
* @package crmeb\swoole\server
*/
trait InteractsWithQueue
{public function prepareQueue(Manager $manager)
{
/** @var QueueManager $queueManager */
$queueManager = $this->laravel->make(QueueManager::class);
$queueManager->attachToServer($manager, $this->output);
}
}Manager 类
<?php
namespace crmeb\swoole\queue;
use Illuminate\Contracts\Container\Container;
use Swoole\Constant;
use Swoole\Process;
use Swoole\Process\Pool;
use Swoole\Timer;
use Illuminate\Support\Arr;
use Illuminate\Queue\Events\JobFailed;
use Illuminate\Queue\Worker;
use crmeb\swoole\server\WithContainer;
use Illuminate\Queue\Jobs\Job;
use function Swoole\Coroutine\run;
use Illuminate\Queue\WorkerOptions;
use SwooleTW\Http\Server\Manager as ServerManager;
use Illuminate\Console\OutputStyle;
class Manager
{
use WithContainer;
/**
* Container.
*
* @var \Illuminate\Contracts\Container\Container
*/
protected $container;
/**
* @var OutputStyle
*/
protected $output;
/**
* @var Closure[]
*/
protected $workers = [];
/**
* Manager constructor.
* @param Container $container
*/
public function __construct(Container $container)
{$this->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("<{$type}>[%s][%s] %s</{$type}> %s",
date('Y-m-d H:i:s'),
$job->getJobId(),
str_pad("{$status}:", 11), $job->getName()));
}
}
减少 CrmebServiceProvider 类
<?php
namespace crmeb\swoole;
use Illuminate\Contracts\Debug\ExceptionHandler;
use Illuminate\Contracts\Http\Kernel;
use crmeb\swoole\command\HttpServerCommand;
use Illuminate\Queue\Worker;
use SwooleTW\Http\HttpServiceProvider;
use SwooleTW\Http\Middleware\AccessLog;
use SwooleTW\Http\Server\Manager;
/**
* Class CrmebServiceProvider
* @package crmeb\swoole
*/
class CrmebServiceProvider extends HttpServiceProvider
{
/**
* Register manager.
*
* @return void
*/
protected function registerManager()
{$this->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');
}
}
而后再把 \crmeb\swoole\CrmebServiceProvider::class 放入 config/app.php 中的 providers 中加载重写了 swoole 的命令启动形式
配置config/swoole_http.php
return [
'queue' => [
// 是否开启主动生产队列
'enabled' => true,
'workers' => [
// 队列名称
'CRMEB' => []]
],
];输出命令:php artisan crmeb:http restart
swoole 启动后就能够主动生产队列了
最初给大哥们举荐个我的项目行行好大哥们点点就行, 感激了:
PC 商城 + 挪动端 uniapp 多端兼容 + 全新架构模式
点链接的大哥们, 一夜暴富, 万事顺心, 春风不倒, 夜夜笙歌
如果你感觉这篇文章对你有点用的话，麻烦请给咱们的开源我的项目点点 star: http://github.crmeb.net/u/defu 不胜感激！
正文完