Yii2融合EasySwoole的消息处理服务

1. 说明:

2. 要求：

3. 实现原理:

4. 实现:

4.1 目录结构

4.2 守护进程服务命令行

4.3 任务

4.4 服务运行示例

1. 说明:

众所周知，Yii2是集优雅与效率于一身的框架，EasySwoole则对Swoole做了人性化的使用包装，两者融合并提供守护应用服务，这里分享我的实践例子。

2. 要求：

1. 在Yii控制台下启动，兼容EasySwoole的种种特性

2. 能正确调取Yii中所有的模型类、组件等等

3. 多进程执行消息处理任务，长时运行稳定

3. 实现原理:

设置守护程序应用(daemon)路径，easyswoole将在这个应用路径下运行
添加 console/EasySwooleController

4. 实现:

4.1 目录结构

daemon为EasySwoole实际运行目录
daemon/dev.php 为easyswoole配置
daemon/log 为easyswoole 运行时日子目录，同时console/runtime/logs为Yii日志目录，兼容

文件修改列表:

src\common\config\main.php - 定义easyswoole根目录

// setup easyswoole root to run task-server
defined('EASYSWOOLE_ROOT') or define('EASYSWOOLE_ROOT', PROJECT_ROOT . '/daemon');

src\daemon\dev.php

<?php/*** refer to https://www.easyswoole.com/QuickStart/config.html*/
return ['SERVER_NAME' => "EasySwoole",'MAIN_SERVER' => ['LISTEN_ADDRESS' => '0.0.0.0','PORT' => 9501,
//        'SERVER_TYPE' => EASYSWOOLE_WEB_SERVER, //可选为 EASYSWOOLE_SERVER  EASYSWOOLE_WEB_SERVER EASYSWOOLE_WEB_SOCKET_SERVER,EASYSWOOLE_REDIS_SERVER'SERVER_TYPE' => EASYSWOOLE_SERVER, //可选为 EASYSWOOLE_SERVER  EASYSWOOLE_WEB_SERVER EASYSWOOLE_WEB_SOCKET_SERVER,EASYSWOOLE_REDIS_SERVER'SOCK_TYPE' => SWOOLE_TCP,'RUN_MODEL' => SWOOLE_PROCESS,'SETTING' => ['worker_num' => 4,'reload_async' => true,'max_wait_time' => 3],'TASK' => ['workerNum' => 4,'maxRunningNum' => 128,'timeout' => 15]],'TEMP_DIR' => '/tmp','LOG' => ['dir' => EASYSWOOLE_ROOT . '/log', //日志文件存放的目录],'clientMqtt1' => [// for dev'clientConfig' => ['userName' => MQTT_DEFAULT_USER, // vhost+username = rabbitmq用户名'password' => RABBITMQ_DEFAULT_PWD, // 密码'clientId' => MQTT_SYS_CLIENT_ID1, // 客户端id1'keepAlive' => 30, // 默认0秒，设置成0代表禁用'protocolName' => 'MQTT', // 协议名，默认为MQTT(3.1.1版本)，也可为MQIsdp(3.1版本)'protocolLevel' => 4, // 协议等级，MQTT3.1.1版本为4，5.0版本为5，MQIsdp为3'properties' => [], // MQTT5 中所需要的属性'delay' => 3000, // 重连时的延迟时间 (毫秒)
//            'maxAttempts' => 5, // 最大重连次数。默认-1，表示不限制'maxAttempts' => -1, // 最大重连次数。默认-1，表示不限制'swooleConfig' => ['open_mqtt_protocol' => true,'package_max_length' => 2 * 1024 * 1024]],'host' => RABBITMQ_DEFAULT_HOST,'port' => MQTT_DEFAULT_PORT,'appID' => 'clientMqtt1','appSecret' => 'public','publishUrl' => 'http://' . RABBITMQ_DEFAULT_HOST . ':15672/api/v4/mqtt/publish',],'clientMqtt2' => [// for app'clientConfig' => ['userName' => MQTT_DEFAULT_USER, // vhost+username = rabbitmq用户名'password' => RABBITMQ_DEFAULT_PWD, // 密码'clientId' => MQTT_SYS_CLIENT_ID2, // 客户端id1'keepAlive' => 30, // 默认0秒，设置成0代表禁用'protocolName' => 'MQTT', // 协议名，默认为MQTT(3.1.1版本)，也可为MQIsdp(3.1版本)'protocolLevel' => 4, // 协议等级，MQTT3.1.1版本为4，5.0版本为5，MQIsdp为3'properties' => [], // MQTT5 中所需要的属性'delay' => 3000, // 重连时的延迟时间 (毫秒)
//            'maxAttempts' => 5, // 最大重连次数。默认-1，表示不限制'maxAttempts' => -1, // 最大重连次数。默认-1，表示不限制'swooleConfig' => ['open_mqtt_protocol' => true,'package_max_length' => 2 * 1024 * 1024]],'host' => RABBITMQ_DEFAULT_HOST,'port' => MQTT_DEFAULT_PORT,'appID' => 'clientMqtt2','appSecret' => 'public','publishUrl' => 'http://' . RABBITMQ_DEFAULT_HOST . ':15672/api/v4/mqtt/publish',],
];

src\console\controllers\EasySwooleController.php

<?phpnamespace console\controllers;use Yii;
use yii\console\Controller;
use yii\helpers\ArrayHelper;
use EasySwoole\EasySwoole\Command\CommandRunner;
use EasySwoole\Command\Caller;/*** Console controller for Easyswoole** @author Ben bi <bennybi@qq.com>*/
class EasySwooleController extends Controller {public $daemon;public $force;public function options($actionID) {return ArrayHelper::merge(parent::options($actionID), ['daemon', 'force',]);}public function optionAliases() {return ArrayHelper::merge(parent::optionAliases(), ['d' => 'daemon','force' => 'force',]);}public function actionServer(...$args) {$argv = $this->getArgv('easy-swoole/server', 'server');
//        Yii::info($argv);$caller = new Caller();$caller->setScript(current($argv));$caller->setCommand(next($argv));$caller->setParams($argv);reset($argv);$ret = CommandRunner::getInstance()->run($caller);if ($ret && !empty($ret->getMsg())) {echo $ret->getMsg() . "\n";}}public function actionProcess(...$args) {$argv = $this->getArgv('easy-swoole/process', 'process');
//        Yii::info($argv);$caller = new Caller();$caller->setScript(current($argv));$caller->setCommand(next($argv));$caller->setParams($argv);reset($argv);$ret = CommandRunner::getInstance()->run($caller);if ($ret && !empty($ret->getMsg())) {echo $ret->getMsg() . "\n";}}public function actionTask(...$args) {$argv = $this->getArgv('easy-swoole/task', 'task');
//        Yii::info($argv);$caller = new Caller();$caller->setScript(current($argv));$caller->setCommand(next($argv));$caller->setParams($argv);reset($argv);$ret = CommandRunner::getInstance()->run($caller);if ($ret && !empty($ret->getMsg())) {echo $ret->getMsg() . "\n";}}protected function getArgv($script, $action) {$argv = $_SERVER['argv'];if (is_array($argv) && $argv['0'] = 'yii') {array_shift($argv);}array_shift($argv);array_unshift($argv, $action);array_unshift($argv, $script);return $argv;}public function getActionHelp($action) {return <<<HELP_START
\e[33m启动操作:\e[0m
\e[31m  php yii easy-swoole/server start [args]\e[0m
\e[33m简介:\e[0m
\e[36m  执行本命令可以启动框架 可选的操作参数如下\e[0m
\e[33m参数:\e[0m
\e[32m  d \e[0m                   以守护模式启动
\e[32m  produce \e[0m             以生产环境启动
\e[33m\e[0m
\e[33m停止操作:\e[0m
\e[31m  php yii easy-swoole/server stop\e[0m
\e[33m\e[0m
\e[33m重启task进程操作:\e[0m
\e[31m  php yii easy-swoole/server reload\e[0m
\e[33m\e[0m
\e[33m重启task + worker进程:\e[0m
\e[31m  php yii easy-swoole/server reload all\e[0m
\e[33m\e[0m
\e[33m进入控制台:\e[0m
\e[31m  php yii easy-swoole/server console\e[0m
HELP_START;}}

daemon/EasSwooleEvent.php 例子

<?phpnamespace EasySwoole\EasySwoole;use EasySwoole\EasySwoole\Swoole\EventRegister;
use EasySwoole\EasySwoole\AbstractInterface\Event;
use EasySwoole\Http\Request;
use EasySwoole\Http\Response;
use EasySwoole\Component\Process\Config;
use EasySwoole\ORM\Db\Connection;
use EasySwoole\ORM\DbManager;
use EasySwoole\Component\Di;
use EasySwoole\Component\Process\Manager;/*** @author Ben bi <bennybi@qq.com>*/
class EasySwooleEvent implements Event {public static function initialize() {date_default_timezone_set('Asia/Shanghai');}public static function mainServerCreate(EventRegister $register) {/*         * * 设备监控进程mqtt (实时) - 自建队列实时监控设备状态 ** */$devMonitorConfig = new Config();$devMonitorConfig->setProcessName(PROJECT_CODE . '.DeviceMonitorRealTime'); //设置进程名称$devMonitorConfig->setProcessGroup(PROJECT_CODE); //设置进程组$devMonitorConfig->setArg([]); //传参$devMonitorConfig->setRedirectStdinStdout(false); //是否重定向标准io$devMonitorConfig->setPipeType($devMonitorConfig::PIPE_TYPE_SOCK_DGRAM); //设置管道类型$devMonitorConfig->setEnableCoroutine(true); //是否自动开启协程$devMonitorConfig->setMaxExitWaitTime(3); //最大退出等待时间Manager::getInstance()->addProcess(new \daemon\process\rabbitmq\mqtt\IotDeviceMonitorRealTime($devMonitorConfig));/*         * * 设备监控进程mqtt (批量) - 监控IA-TE001-DEVICE-STATUS 持久队列, 防止漏掉信息** */Manager::getInstance()->addProcess(new \daemon\process\rabbitmq\mqtt\IotDeviceMonitorBatch(PROJECT_CODE . '.DeviceMonitorBatch'));/*         * * 传感器Sensors消息进程mqtt (批量) ** */Manager::getInstance()->addProcess(new \daemon\process\rabbitmq\mqtt\IotSensorReportMonitorBatch(PROJECT_CODE . '.SensorReportMonitorBatch'));/*         * * 解析原始数据 (批量) ** */Manager::getInstance()->addProcess(new \daemon\process\rabbitmq\mqtt\IotRawdataParserBatch(PROJECT_CODE . '.RawdataParserBatch'));$register->add($register::onConnect, function (\Swoole\Server $server, int $fd, int $reactor_id) {echo "fd {$fd} connected";});$register->add($register::onReceive, function (\Swoole\Server $server, int $fd, int $reactor_id, string $data) {echo "fd:{$fd} send:{$data}\n";});$register->add($register::onClose, function (\Swoole\Server $server, int $fd, int $reactor_id) {echo "fd {$fd} closed";});}public static function onRequest(Request $request, Response $response): bool {// TODO: Implement onRequest() method.return true;}public static function afterRequest(Request $request, Response $response): void {// TODO: Implement afterAction() method.}}

4.2 守护进程服务命令行

命令	备注
Docker 方式
docker exec ia-php php /var/www/html/te001/yii easy-swoole/server start -d
docker exec ia-php php /var/www/html/te001/yii easy-swoole/server stop

一般方式
php yii easy-swoole/server start -d	守护模式启动
php yii easy-swoole/server start	开发模式启动
php yii easy-swoole/server stop	停止
php yii easy-swoole/server restart	重启

4.3 任务

\daemon\process\rabbitmq\mqtt\IotDeviceMonitorBatch 设备监控进程mqtt (批量) - 监控IA-TE001-DEVICE-STATUS 持久队列

\daemon\process\rabbitmq\mqtt\IotSensorReportMonitorBatch 传感器Sensors消息进程mqtt (批量)

\daemon\process\rabbitmq\mqtt\IotRawdataParserBatch 解析原始数据 (批量)

4.4 服务运行示例