1

app/command/Workerman.php

@@ -0,0 +1,98 @@
+<?php
+// +----------------------------------------------------------------------
+// | Workerman 命令行入口
+// +----------------------------------------------------------------------
+namespace app\command;
+
+use Channel\Server;
+use think\console\Command;
+use think\console\Input;
+use think\console\input\Argument;
+use think\console\input\Option;
+use think\console\Output;
+use Workerman\Worker;
+
+class Workerman extends Command
+{
+    protected $config = [];
+    protected $workerServer;
+    protected $chatWorkerServer;
+    protected $channelServer;
+
+    protected function configure()
+    {
+        $this->setName('workerman')
+            ->addArgument('status', Argument::REQUIRED, 'start/stop/reload/status')
+            ->addArgument('server', Argument::OPTIONAL, 'admin/chat/channel')
+            ->addOption('d', null, Option::VALUE_NONE, '守护进程方式启动')
+            ->setDescription('Workerman服务管理命令');
+    }
+
+    protected function execute(Input $input, Output $output)
+    {
+        global $argv;
+        $argv[1] = $input->getArgument('status') ?: 'start';
+        $server = $input->getArgument('server');
+        
+        if ($input->hasOption('d')) {
+            $argv[2] = '-d';
+        } else {
+            unset($argv[2]);
+        }
+
+        $this->config = config('workerman');
+        Worker::$pidFile = app()->getRootPath() . 'runtime/workerman.pid';
+
+        if (!$server || $server == 'admin') {
+            $output->info('启动 Admin 服务: 0.0.0.0:' . $this->config['admin']['port']);
+            $this->workerServer = new Worker(
+                $this->config['admin']['protocol'] . '://' . 
+                $this->config['admin']['ip'] . ':' . 
+                $this->config['admin']['port']
+            );
+            $this->workerServer->count = $this->config['admin']['serverCount'];
+        }
+
+        if (!$server || $server == 'chat') {
+            $output->info('启动 Chat 服务: 0.0.0.0:' . $this->config['chat']['port']);
+            $this->chatWorkerServer = new Worker(
+                $this->config['chat']['protocol'] . '://' . 
+                $this->config['chat']['ip'] . ':' . 
+                $this->config['chat']['port']
+            );
+            $this->chatWorkerServer->count = $this->config['chat']['serverCount'];
+        }
+
+        if (!$server || $server == 'channel') {
+            $output->info('启动 Channel 服务: ' . $this->config['channel']['ip'] . ':' . $this->config['channel']['port']);
+            $this->channelServer = new Server($this->config['channel']['ip'], $this->config['channel']['port']);
+        }
+
+        $this->bindHandle();
+        
+        try {
+            Worker::runAll();
+        } catch (\Exception $e) {
+            $output->warning($e->getMessage());
+        }
+    }
+
+    protected function bindHandle()
+    {
+        if (!is_null($this->workerServer)) {
+            $server = new \app\services\workerman\WorkermanService($this->workerServer, $this->channelServer);
+            $this->workerServer->onConnect = [$server, 'onConnect'];
+            $this->workerServer->onMessage = [$server, 'onMessage'];
+            $this->workerServer->onWorkerStart = [$server, 'onWorkerStart'];
+            $this->workerServer->onClose = [$server, 'onClose'];
+        }
+
+        if (!is_null($this->chatWorkerServer)) {
+            $chatServer = new \app\services\workerman\chat\ChatService($this->chatWorkerServer, $this->channelServer);
+            $this->chatWorkerServer->onConnect = [$chatServer, 'onConnect'];
+            $this->chatWorkerServer->onMessage = [$chatServer, 'onMessage'];
+            $this->chatWorkerServer->onWorkerStart = [$chatServer, 'onWorkerStart'];
+            $this->chatWorkerServer->onClose = [$chatServer, 'onClose'];
+        }
+    }
+}

app/services/workerman/ChannelService.php

@@ -0,0 +1,52 @@
+<?php
+// +----------------------------------------------------------------------
+// | Channel 进程通信服务
+// +----------------------------------------------------------------------
+namespace app\services\workerman;
+
+use Channel\Client;
+
+class ChannelService
+{
+    protected $trigger = 'muyinjie';
+    protected static $instance;
+
+    public function __construct()
+    {
+        self::connet();
+    }
+
+    public static function instance()
+    {
+        if (is_null(self::$instance)) {
+            self::$instance = new self();
+        }
+        return self::$instance;
+    }
+
+    public static function connet()
+    {
+        $config = config('workerman.channel');
+        Client::connect($config['ip'], $config['port']);
+    }
+
+    public function setTrigger(string $name)
+    {
+        $this->trigger = $name;
+        return $this;
+    }
+
+    public function send(string $type, ?array $data = null, ?array $ids = null)
+    {
+        $res = compact('type');
+        if (!is_null($data)) $res['data'] = $data;
+        if (!is_null($ids) && count($ids)) $res['ids'] = $ids;
+        $this->trigger($this->trigger, $res);
+        $this->trigger = 'muyinjie';
+    }
+
+    public function trigger(string $type, ?array $data = null)
+    {
+        Client::publish($type, $data);
+    }
+}

app/services/workerman/Response.php

@@ -0,0 +1,56 @@
+<?php
+// +----------------------------------------------------------------------
+// | Workerman 响应封装类
+// +----------------------------------------------------------------------
+namespace app\services\workerman;
+
+use Workerman\Connection\TcpConnection;
+
+class Response
+{
+    protected $connection;
+
+    public function connection(TcpConnection $connection)
+    {
+        $this->connection = $connection;
+        return $this;
+    }
+
+    public function send(string $type, ?array $data = null, bool $close = false, array $other = [])
+    {
+        $this->connection->lastMessageTime = time();
+        $res = compact('type');
+        if (!is_null($data)) $res['data'] = $data;
+        $data = array_merge($res, $other);
+        if ($close) $data['close'] = true;
+        $json = json_encode($data);
+        return $close ? $this->connection->close($json) : $this->connection->send($json);
+    }
+
+    public function success($type = 'success', ?array $data = null)
+    {
+        if (is_array($type)) {
+            $data = $type;
+            $type = 'success';
+        }
+        return $this->send($type, $data);
+    }
+
+    public function fail($type = 'error', ?array $data = null)
+    {
+        if (is_array($type)) {
+            $data = $type;
+            $type = 'error';
+        }
+        return $this->send($type, $data);
+    }
+
+    public function close($type = 'error', ?array $data = null)
+    {
+        if (is_array($type)) {
+            $data = $type;
+            $type = 'error';
+        }
+        return $this->send($type, $data, true);
+    }
+}

app/services/workerman/WorkermanHandle.php

@@ -0,0 +1,46 @@
+<?php
+// +----------------------------------------------------------------------
+// | Admin 消息处理类
+// +----------------------------------------------------------------------
+namespace app\services\workerman;
+
+use Workerman\Connection\TcpConnection;
+
+class WorkermanHandle
+{
+    protected $service;
+
+    public function __construct(WorkermanService &$service)
+    {
+        $this->service = &$service;
+    }
+
+    /**
+     * 管理员登录
+     */
+    public function login(TcpConnection &$connection, array $res, Response $response)
+    {
+        if (!isset($res['data']) || !$token = $res['data']) {
+            return $response->close(['msg' => '授权失败!']);
+        }
+
+        // TODO: 根据你的系统实现token验证
+        // 示例：$authInfo = your_auth_service::parseToken($token);
+        
+        // 临时示例：直接接受登录
+        $authInfo = [
+            'id' => $res['data']['id'] ?? 1,
+            'name' => $res['data']['name'] ?? 'Admin',
+        ];
+
+        if (!$authInfo || !isset($authInfo['id'])) {
+            return $response->close(['msg' => '授权失败!']);
+        }
+
+        $connection->adminInfo = $authInfo;
+        $connection->adminId = $authInfo['id'];
+        $this->service->setUser($connection);
+
+        return $response->success();
+    }
+}

app/services/workerman/WorkermanService.php

@@ -0,0 +1,81 @@
+<?php
+// +----------------------------------------------------------------------
+// | Admin 通知服务处理类
+// +----------------------------------------------------------------------
+namespace app\services\workerman;
+
+use Channel\Client;
+use Workerman\Connection\TcpConnection;
+use Workerman\Lib\Timer;
+use Workerman\Worker;
+
+class WorkermanService
+{
+    protected $worker;
+    protected $connections = [];
+    protected $user = [];
+    protected $handle;
+    protected $response;
+    protected $timer;
+
+    public function __construct(Worker $worker)
+    {
+        $this->worker = $worker;
+        $this->handle = new WorkermanHandle($this);
+        $this->response = new Response();
+    }
+
+    public function setUser(TcpConnection $connection)
+    {
+        $this->user[$connection->adminInfo['id']] = $connection;
+    }
+
+    public function onConnect(TcpConnection $connection)
+    {
+        $this->connections[$connection->id] = $connection;
+        $connection->lastMessageTime = time();
+    }
+
+    public function onMessage(TcpConnection $connection, $res)
+    {
+        $connection->lastMessageTime = time();
+        $res = json_decode($res, true);
+        if (!$res || !isset($res['type']) || !$res['type'] || $res['type'] == 'ping') {
+            return $this->response->connection($connection)->success('ping', ['now' => time()]);
+        }
+        if (!method_exists($this->handle, $res['type'])) return;
+        $this->handle->{$res['type']}($connection, $res + ['data' => []], $this->response->connection($connection));
+    }
+
+    public function onWorkerStart(Worker $worker)
+    {
+        ChannelService::connet();
+
+        Client::on('muyinjie', function ($eventData) use ($worker) {
+            if (!isset($eventData['type']) || !$eventData['type']) return;
+            $ids = isset($eventData['ids']) && count($eventData['ids']) ? $eventData['ids'] : array_keys($this->user);
+            foreach ($ids as $id) {
+                if (isset($this->user[$id])) {
+                    $this->response->connection($this->user[$id])->success($eventData['type'], $eventData['data'] ?? null);
+                }
+            }
+        });
+
+        $this->timer = Timer::add(15, function () use (&$worker) {
+            $time_now = time();
+            foreach ($worker->connections as $connection) {
+                if ($time_now - $connection->lastMessageTime > 12) {
+                    $this->response->connection($connection)->close('timeout');
+                }
+            }
+        });
+    }
+
+    public function onClose(TcpConnection $connection)
+    {
+        unset($this->connections[$connection->id]);
+        if (isset($connection->adminId)) {
+            unset($this->user[$connection->adminId]);
+        }
+    }
+}

app/services/workerman/chat/ChatHandle.php

@@ -0,0 +1,106 @@
+<?php
+// +----------------------------------------------------------------------
+// | Chat 消息处理类
+// +----------------------------------------------------------------------
+namespace app\services\workerman\chat;
+
+use app\services\workerman\Response;
+use Workerman\Connection\TcpConnection;
+
+class ChatHandle
+{
+    protected $service;
+
+    public function __construct(ChatService &$service)
+    {
+        $this->service = &$service;
+    }
+
+    /**
+     * 客服登录
+     */
+    public function kefu_login(TcpConnection &$connection, array $res, Response $response)
+    {
+        if (!isset($res['data']) || !$token = $res['data']) {
+            return $response->close(['msg' => '授权失败!']);
+        }
+
+        // TODO: 根据你的系统实现客服token验证
+        // 示例：$kefuInfo = your_kefu_auth_service::parseToken($token);
+        
+        // 临时示例
+        $kefuInfo = [
+            'uid' => $res['data']['uid'] ?? 1,
+            'name' => $res['data']['name'] ?? '客服',
+        ];
+
+        $connection->kefuUser = (object)$kefuInfo;
+        $connection->user = (object)['uid' => $kefuInfo['uid'], 'nickname' => $kefuInfo['name'] ?? '客服'];
+        $this->service->setKefuUser($connection);
+
+        return $response->success();
+    }
+
+    /**
+     * 用户登录
+     */
+    public function login(TcpConnection &$connection, array $res, Response $response)
+    {
+        if (!isset($res['data']) || !$token = $res['data']) {
+            return $response->close(['msg' => '授权失败!']);
+        }
+
+        // TODO: 根据你的系统实现用户token验证
+        // 示例：$authInfo = your_user_auth_service::parseToken($token);
+        
+        // 临时示例
+        $userInfo = [
+            'uid' => $res['data']['uid'] ?? 1,
+            'nickname' => $res['data']['nickname'] ?? '用户',
+            'avatar' => $res['data']['avatar'] ?? '',
+        ];
+
+        $connection->user = (object)$userInfo;
+        $this->service->setUser($connection);
+
+        return $response->success();
+    }
+
+    /**
+     * 发送消息
+     */
+    public function chat(TcpConnection &$connection, array $res, Response $response)
+    {
+        $to_uid = $res['data']['to_uid'] ?? 0;
+        $msn = $res['data']['msn'] ?? '';
+        $msn_type = $res['data']['type'] ?? 1; // 1=文本, 2=图片, 3=语音
+
+        if (!$to_uid) {
+            return $response->send('err_tip', ['msg' => '用户不存在']);
+        }
+
+        if ($to_uid == ($connection->user->uid ?? 0)) {
+            return $response->send('err_tip', ['msg' => '不能和自己聊天']);
+        }
+
+        $uid = $connection->user->uid ?? 0;
+        
+        $data = [
+            'from_uid' => $uid,
+            'to_uid' => $to_uid,
+            'msn' => $msn,
+            'type' => $msn_type,
+            'add_time' => time(),
+            'nickname' => $connection->user->nickname ?? '用户',
+            'avatar' => $connection->user->avatar ?? '',
+        ];
+
+        // 给自己回复
+        $response->send('chat', $data);
+
+        // 发送给目标用户
+        if (isset($this->service->user()[$to_uid])) {
+            $this->response->connection($this->service->user()[$to_uid])->send('reply', $data);
+        }
+    }
+}

app/services/workerman/chat/ChatService.php

@@ -0,0 +1,109 @@
+<?php
+// +----------------------------------------------------------------------
+// | Chat 客服聊天服务
+// +----------------------------------------------------------------------
+namespace app\services\workerman\chat;
+
+use Channel\Client;
+use app\services\workerman\ChannelService;
+use app\services\workerman\Response;
+use Workerman\Connection\TcpConnection;
+use Workerman\Lib\Timer;
+use Workerman\Worker;
+
+class ChatService
+{
+    protected $worker;
+    protected $connections = [];
+    protected $user = [];
+    protected $kefuUser = [];
+    protected $handle;
+    protected $response;
+    protected $timer;
+
+    public function __construct(Worker $worker)
+    {
+        $this->worker = $worker;
+        $this->handle = new ChatHandle($this);
+        $this->response = new Response();
+    }
+
+    public function setUser(TcpConnection $connection)
+    {
+        $this->user[$connection->user->uid] = $connection;
+    }
+
+    public function kefuUser()
+    {
+        return $this->kefuUser;
+    }
+
+    public function setKefuUser(TcpConnection $connection, bool $isUser = true)
+    {
+        $this->kefuUser[$connection->kefuUser->uid] = $connection;
+        if ($isUser) {
+            $this->user[$connection->user->uid] = $connection;
+        }
+    }
+
+    public function user($key = null)
+    {
+        return $key ? ($this->user[$key] ?? false) : $this->user;
+    }
+
+    public function onConnect(TcpConnection $connection)
+    {
+        $this->connections[$connection->id] = $connection;
+        $connection->lastMessageTime = time();
+    }
+
+    public function onMessage(TcpConnection $connection, $res)
+    {
+        $connection->lastMessageTime = time();
+        $res = json_decode($res, true);
+        if (!$res || !isset($res['type']) || !$res['type'] || $res['type'] == 'ping') {
+            return $this->response->connection($connection)->success('ping', ['now' => time()]);
+        }
+        if (!method_exists($this->handle, $res['type'])) return;
+        try {
+            $this->handle->{$res['type']}($connection, $res + ['data' => []], $this->response->connection($connection));
+        } catch (\Throwable $e) {
+            // 忽略异常
+        }
+    }
+
+    public function onWorkerStart(Worker $worker)
+    {
+        ChannelService::connet();
+
+        Client::on('muyinjie_chat', function ($eventData) use ($worker) {
+            if (!isset($eventData['type']) || !$eventData['type']) return;
+            $ids = isset($eventData['ids']) && count($eventData['ids']) ? $eventData['ids'] : array_keys($this->user);
+            foreach ($ids as $id) {
+                if (isset($this->user[$id])) {
+                    $this->response->connection($this->user[$id])->success($eventData['type'], $eventData['data'] ?? null);
+                }
+            }
+        });
+
+        $this->timer = Timer::add(15, function () use (&$worker) {
+            $time_now = time();
+            foreach ($worker->connections as $connection) {
+                if ($time_now - $connection->lastMessageTime > 12) {
+                    $this->response->connection($connection)->close('timeout');
+                }
+            }
+        });
+    }
+
+    public function onClose(TcpConnection $connection)
+    {
+        unset($this->connections[$connection->id]);
+        if (isset($connection->user->uid)) {
+            unset($this->user[$connection->user->uid]);
+        }
+        if (isset($connection->kefuUser->uid)) {
+            unset($this->kefuUser[$connection->kefuUser->uid]);
+        }
+    }
+}

config/workerman.php

@@ -0,0 +1,28 @@
+<?php
+// +----------------------------------------------------------------------
+// | Workerman配置文件
+// +----------------------------------------------------------------------
+
+return [
+    // 内部通讯端口（进程间通信）
+    'channel' => [
+        'port' => 20220,
+        'ip' => '127.0.0.1',
+    ],
+
+    // Admin 后台通知服务
+    'admin' => [
+        'protocol' => 'websocket',
+        'ip' => '0.0.0.0',
+        'port' => 20221,
+        'serverCount' => 1,
+    ],
+
+    // Chat 客服聊天服务
+    'chat' => [
+        'protocol' => 'websocket',
+        'ip' => '0.0.0.0',
+        'port' => 20222,
+        'serverCount' => 1,
+    ],
+];