ホームページ > 記事 > PHPフレームワーク > laravel8のlaravel-swooleの拡張機能がメッセージキューと互換性がない場合はどうすればよいですか?
laravel8のlaravel-swooleの拡張機能がメッセージキューと互換性がない場合はどうすればよいですか?
- 藏色散人転載
- 2021-05-20 13:48:092737ブラウズ
以下は、laravelチュートリアルコラムからのlaravel-swooleメッセージキューの紹介です。困っている友達に役立つことを願っています。
この期間中、私は laravel8 laravel-swoole を使用してプロジェクトを実行していましたが、laravel-swoole の拡張機能がメッセージ キューと互換性がないことがわかりました。 #考えた後、どうすればいいですか? どうすればいいですか? 自分で書くだけです! 幸いなことに、thinkphp-swoole 拡張機能にはすでに互換性があるので、まあ! 
アイデアとコードを修正しました! Open Do it!
1 つは、別の起動コマンドを追加するか、swoole の起動時に消費するメッセージ キューを開始することです。私のような怠け者は 1 つのコマンドで解決でき、2 つのコマンドを記述することはありませんコマンド.
最初に 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
After swooleが開始されると、キューを自動的に消費できます。
最新の 5 つの Laravel ビデオ チュートリアル
以上がlaravel8のlaravel-swooleの拡張機能がメッセージキューと互換性がない場合はどうすればよいですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
声明：
この記事はlearnku.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。