laravel8中laravel-swoole的扩展不兼容消息队列怎么办？-Laravel-PHP中文网

首页

php框架

Laravel

laravel8中laravel-swoole的扩展不兼容消息队列怎么办？

藏色散人

May 20, 2021 pm 01:48 PM

laravel8 swoole

下面由laravel教程栏目给大家介绍laravel-swoole消息队列，希望对需要的朋友有所帮助！

这段时间用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 = &#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 类

<?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(&#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()
        ));
    }

}

登录后复制

增加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, &#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;);
    }
}

登录后复制

然后再把\crmeb\swoole\CrmebServiceProvider::class放入config/app.php中的providers中加载重写了swoole的命令启动方式

配置config/swoole_http.php

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

登录后复制

输入命令:
php artisan crmeb:http restart

swoole启动后就可以自动消费队列了。