What should I do if the extension of laravel-swoole in laravel8 is not compatible with the message queue?

The following is an introduction to laravel-swoole message queue from the laraveltutorial column, I hope it will be helpful to friends in need!

During this period of time, I used laravel8 laravel-swoole to do projects, and found that the extension of laravel-swoole is not compatible with the message queue;

After thinking about it, what should I do? What should I do? Just write it yourself! Fortunately, the thinkphp-swoole extension is already compatible, so hey!

Directly upload the modified ideas and code! Open Do it!

One is to add another startup command or start the message queue for consumption when swoole is started. A lazy person like me can solve it with one command and will never write two commands.

First rewrite the swoole startup command

<?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 = &#39;crmeb:http {action : start|stop|restart|reload|infos}&#39;;

    /**
     * Run swoole_http_server.
     */
    protected function start()
    {
        if ($this->isRunning()) {
            $this->error(&#39;Failed! swoole_http_server process is already running.&#39;);

            return;
        }

        $host             = Arr::get($this->config, &#39;server.host&#39;);
        $port             = Arr::get($this->config, &#39;server.port&#39;);
        $hotReloadEnabled = Arr::get($this->config, &#39;hot_reload.enabled&#39;);
        $queueEnabled     = Arr::get($this->config, &#39;queue.enabled&#39;);
        $accessLogEnabled = Arr::get($this->config, &#39;server.access_log&#39;);
        $coroutineEnable  = Arr::get($this->config, &#39;coroutine.enable&#39;);

        $this->info(&#39;Starting swoole http server...&#39;);
        $this->info("Swoole http server started: <http://{$host}:{$port}>");
        if ($this->isDaemon()) {
            $this->info(
                &#39;> (You can run this command to ensure the &#39; .
                &#39;swoole_http_server process is running: ps aux|grep "swoole")&#39;
            );
        }

        $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, &#39;coroutine.flags&#39;, 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, &#39;hot_reload.include&#39;, []),
                Arr::get($this->config, &#39;hot_reload.exclude&#39;, []),
                Arr::get($this->config, &#39;hot_reload.name&#39;, [])
            );

            $watcher->watch(function () use ($server) {
                $server->reload();
            });
        }, false, 0, true);
    }

}

InteractsWithQueue class

<?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 class

<?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(&#39;queue.workers&#39;, []);

        foreach ($workers as $queue => $options) {

            if (strpos($queue, &#39;@&#39;) !== false) {
                [$queue, $connection] = explode(&#39;@&#39;, $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(&#39;queue.worker&#39;);
                /** @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(&#39;events&#39;)->listen(JobFailed::class, function (JobFailed $event) {
            $this->writeOutput($event->job);

            $this->logFailedJob($event);
        });
    }

    /**
     * 记录失败任务
     * @param JobFailed $event
     */
    protected function logFailedJob(JobFailed $event)
    {
        $this->container[&#39;queue.failer&#39;]->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 &#39;starting&#39;:
                $this->writeStatus($job, &#39;Processing&#39;, &#39;comment&#39;);
                break;
            case &#39;success&#39;:
                $this->writeStatus($job, &#39;Processed&#39;, &#39;info&#39;);
                break;
            case &#39;failed&#39;:
                $this->writeStatus($job, &#39;Failed&#39;, &#39;error&#39;);
                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(&#39;Y-m-d H:i:s&#39;),
            $job->getJobId(),
            str_pad("{$status}:", 11), $job->getName()
        ));
    }

}

Add CrmebServiceProvider class

<?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, &#39;laravel&#39;);
        });

        $this->app->alias(Manager::class, &#39;swoole.manager&#39;);

        $this->app->singleton(&#39;queue.worker&#39;, function ($app) {
            $isDownForMaintenance = function () {
                return $this->app->isDownForMaintenance();
            };

            return new Worker(
                $app[&#39;queue&#39;],
                $app[&#39;events&#39;],
                $app[ExceptionHandler::class],
                $isDownForMaintenance
            );
        });
    }

    /**
     * Boot websocket routes.
     *
     * @return void
     */
    protected function bootWebsocketRoutes()
    {
        require base_path(&#39;vendor/swooletw/laravel-swoole&#39;) . &#39;/routes/laravel_routes.php&#39;;
    }

    /**
     * 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(&#39;vendor/swooletw/laravel-swoole&#39;) . &#39;/config/swoole_http.php&#39;, &#39;swoole_http&#39;);
        $this->mergeConfigFrom(base_path(&#39;vendor/swooletw/laravel-swoole&#39;) . &#39;/config/swoole_websocket.php&#39;, &#39;swoole_websocket&#39;);
    }

    /**
     * Publish files of this package.
     */
    protected function publishFiles()
    {
        $this->publishes([
            base_path(&#39;vendor/swooletw/laravel-swoole&#39;) . &#39;/config/swoole_http.php&#39; => base_path(&#39;config/swoole_http.php&#39;),
            base_path(&#39;vendor/swooletw/laravel-swoole&#39;) . &#39;/config/swoole_websocket.php&#39; => base_path(&#39;config/swoole_websocket.php&#39;),
            base_path(&#39;vendor/swooletw/laravel-swoole&#39;) . &#39;/routes/websocket.php&#39; => base_path(&#39;routes/websocket.php&#39;),
        ], &#39;laravel-swoole&#39;);
    }
}

Then add \ crmeb\swoole\CrmebServiceProvider::classPut it into providers in config/app.php to load and rewrite the swoole command startup method

configuration config/swoole_http.php

return [
    'queue'        => [
        //是否开启自动消费队列
        'enabled' => true,
        'workers' => [
            //队列名称
            'CRMEB' => []
        ]
    ],];

Enter the command:
php artisan crmeb:http restart

After swoole is started, you can automatically consume the queue .

Related recommendations: The latest five Laravel video tutorials

The above is the detailed content of What should I do if the extension of laravel-swoole in laravel8 is not compatible with the message queue?. For more information, please follow other related articles on the PHP Chinese website!