튜토리얼 칼럼에서 Laravel-Swoole 메시지 큐를 소개한 내용인데, 필요한 친구들에게 도움이 되었으면 좋겠습니다! 이 기간 동안 laravel8+laravel-swoole을 사용하여 프로젝트를 수행했는데 laravel-swoole의 확장이 메시지 대기열과 호환되지 않는다는 것을 발견했습니다.
생각해 보니 어떻게 해야 할까요? 어떻게 해야 할까요? 그럼 직접 작성해 보세요! 다행스럽게도 thinkphp-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);
}
}
<?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);
}
}<?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()
));
}
}config/app.php의 providers에 crmebswooleCrmebServiceProvider::class를 넣어 로드합니다. swoole 명령 시작 방법을 다시 작성하세요 구성config/swoole_http.php<?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');
}
}php artisan crmeb:http restart
Swoole은 자동으로 시작된 후 대기열에 넣습니다.
관련 추천:최신 5개의 Laravel 비디오 튜토리얼crmebswooleCrmebServiceProvider::class放入config/app.php中的providers中加载重写了swoole的命令启动方式
配置config/swoole_http.php
return [
'queue' => [
//是否开启自动消费队列
'enabled' => true,
'workers' => [
//队列名称
'CRMEB' => []
]
],];输入命令:php artisan crmeb:http restart
위 내용은 laravel8의 laravel-swoole 확장이 메시지 대기열과 호환되지 않으면 어떻게 해야 합니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
