ThinkPHP项目Workerman长连接应用

wen PHP项目 2026-08-12 168

本文目录导读：

环境准备
创建业务处理类
命令控制器
客户端实现
ThinkPHP集成配置
启动和管理服务
高级功能扩展
性能优化建议
注意事项

我来详细说明如何在ThinkPHP项目中使用Workerman实现长连接应用。

环境准备

安装Workerman

composer require workerman/workerman

创建Workerman服务类

创建 app/worker/WorkerServer.php：

<?php
namespace app\worker;
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use think\facade\Log;
class WorkerServer
{
    private $worker;
    public function __construct()
    {
        // 创建一个Worker监听端口
        $this->worker = new Worker("websocket://0.0.0.0:2346");
        $this->worker->count = 4; // 启动4个进程
        $this->worker->name = 'ChatServer';
        // 设置回调
        $this->worker->onConnect = [$this, 'onConnect'];
        $this->worker->onMessage = [$this, 'onMessage'];
        $this->worker->onClose = [$this, 'onClose'];
        $this->worker->onError = [$this, 'onError'];
        $this->worker->onWorkerStart = [$this, 'onWorkerStart'];
    }
    /**
     * 进程启动时执行
     */
    public function onWorkerStart($worker)
    {
        // 初始化数据库连接等
        echo "Worker started\n";
    }
    /**
     * 客户端连接时
     */
    public function onConnect(TcpConnection $connection)
    {
        echo "New connection from: {$connection->getRemoteIp()}\n";
    }
    /**
     * 收到消息时
     */
    public function onMessage(TcpConnection $connection, $data)
    {
        $data = json_decode($data, true);
        // 处理业务逻辑
        $response = $this->handleBusiness($data, $connection);
        // 发送响应
        $connection->send(json_encode($response));
    }
    /**
     * 处理具体业务
     */
    private function handleBusiness($data, $connection)
    {
        // 通过ThinkPHP的业务逻辑处理
        try {
            // 这里可以调用业务逻辑类
            $service = new \app\service\BusinessService();
            $result = $service->process($data);
            return ['code' => 200, 'msg' => 'success', 'data' => $result];
        } catch (\Exception $e) {
            Log::error('Worker业务处理失败: ' . $e->getMessage());
            return ['code' => 500, 'msg' => '处理失败', 'error' => $e->getMessage()];
        }
    }
    /**
     * 连接关闭时
     */
    public function onClose(TcpConnection $connection)
    {
        echo "Connection closed\n";
    }
    /**
     * 错误处理
     */
    public function onError(TcpConnection $connection, $code, $msg)
    {
        echo "Error: $code - $msg\n";
        Log::error("Workerman error: $code - $msg");
    }
    /**
     * 运行服务
     */
    public function start()
    {
        Worker::runAll();
    }
}

创建业务处理类

创建 app/service/BusinessService.php：

<?php
namespace app\service;
use think\facade\Db;
use think\facade\Cache;
class BusinessService
{
    /**
     * 处理业务逻辑
     */
    public function process($data)
    {
        $type = $data['type'] ?? '';
        switch ($type) {
            case 'login':
                return $this->handleLogin($data);
            case 'chat':
                return $this->handleChat($data);
            case 'heartbeat':
                return $this->handleHeartbeat();
            default:
                return ['error' => '未知的业务类型'];
        }
    }
    /**
     * 处理登录
     */
    private function handleLogin($data)
    {
        $userId = $data['user_id'] ?? 0;
        // 更新用户在线状态
        Cache::set("user_online_{$userId}", 1, 3600);
        return "用户 {$userId} 登录成功";
    }
    /**
     * 处理聊天
     */
    private function handleChat($data)
    {
        $message = $data['message'] ?? '';
        $userId = $data['user_id'] ?? 0;
        // 处理聊天逻辑
        return "用户 {$userId} 说: {$message}";
    }
    /**
     * 处理心跳
     */
    private function handleHeartbeat()
    {
        return 'pong';
    }
}

命令控制器

创建 app/command/WorkerCommand.php：

<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use app\worker\WorkerServer;
class WorkerCommand extends Command
{
    protected function configure()
    {
        $this->setName('workerman')
            ->setDescription('Workerman 服务控制器')
            ->addArgument('action', 'start|stop|restart|reload|status');
    }
    protected function execute(Input $input, Output $output)
    {
        $action = $input->getArgument('action');
        // 设置全局配置
        global $argv;
        $argv = ['think', $action];
        $server = new WorkerServer();
        $server->start();
    }
}

客户端实现

创建前端JavaScript部分：

// WebSocket客户端示例
class WebSocketClient {
    constructor(url) {
        this.url = url;
        this.ws = null;
        this.isConnected = false;
        this.reconnectTimes = 0;
        this.maxReconnect = 5;
    }
    connect() {
        this.ws = new WebSocket(this.url);
        this.ws.onopen = (e) => {
            console.log('WebSocket连接成功');
            this.isConnected = true;
            this.reconnectTimes = 0;
            this.onConnect && this.onConnect(e);
        };
        this.ws.onmessage = (e) => {
            const data = JSON.parse(e.data);
            this.onMessage && this.onMessage(data);
        };
        this.ws.onclose = (e) => {
            console.log('WebSocket关闭');
            this.isConnected = false;
            this.onClose && this.onClose(e);
            this.reconnect();
        };
        this.ws.onerror = (e) => {
            console.error('WebSocket错误:', e);
            this.onError && this.onError(e);
        };
    }
    send(data) {
        if (this.isConnected) {
            this.ws.send(JSON.stringify(data));
        }
    }
    reconnect() {
        if (this.reconnectTimes < this.maxReconnect) {
            this.reconnectTimes++;
            console.log(`尝试重新连接... (${this.reconnectTimes}/${this.maxReconnect})`);
            setTimeout(() => this.connect(), 3000);
        }
    }
    close() {
        this.isConnected = false;
        this.ws && this.ws.close();
    }
    // 心跳检测
    heartbeat() {
        setInterval(() => {
            if (this.isConnected) {
                this.send({ type: 'heartbeat' });
            }
        }, 30000);
    }
}
// 使用示例
const client = new WebSocketClient('ws://localhost:2346');
client.onMessage = (data) => {
    console.log('收到消息:', data);
    // 处理消息
};
client.connect();
client.heartbeat();

ThinkPHP集成配置

添加Worker配置文件

创建 config/worker.php：

<?php
return [
    // Worker端口
    'port' => 2346,
    // 协议类型
    'protocol' => 'websocket',
    // 进程数
    'worker_count' => 4,
    // 是否启用SSL
    'ssl' => false,
    // 自定义进程列表
    'process' => [],
    // 网关配置
    'gateway' => [],
];

创建入口文件

创建 bin/workerman.php：

#!/usr/bin/env php
<?php
// 加载ThinkPHP框架
require __DIR__ . '/../vendor/autoload.php';
// 设置运行环境
$app = new \think\App();
$app->initialize();
// 启动Workerman
use Workerman\Worker;
use app\worker\WorkerServer;
// 检查扩展
if (!extension_loaded('pcntl')) {
    echo "请安装pcntl扩展\n";
    exit;
}
// 启动服务
$server = new WorkerServer();
Worker::$daemonize = true; // 是否以守护进程方式运行
$server->start();

启动和管理服务

启动服务

# 前台运行
php bin/workerman.php start
# 后台运行
php bin/workerman.php start -d
# 通过ThinkPHP命令
php think workerman

停止服务

php bin/workerman.php stop

重载服务

php bin/workerman.php reload

查看状态

php bin/workerman.php status

高级功能扩展

多控制器支持

class MultiConnectionController
{
    public function handle($connection, $data)
    {
        // 根据消息路由到不同的处理方法
        $route = $data['route'] ?? 'default';
        switch ($route) {
            case 'main':
                return $this->handleMain($connection, $data);
            case 'auxiliary':
                return $this->handleAuxiliary($connection, $data);
            default:
                return ['error' => '无效路由'];
        }
    }
}

消息推送

// 推送给所有客户端
function sendToAll($data)
{
    foreach ($GLOBALS['connections'] as $connection) {
        $connection->send(json_encode($data));
    }
}

定时任务

public function onWorkerStart($worker)
{
    // 定时发送心跳
    \Workerman\Timer::add(5, function() use ($worker) {
        echo "Heartbeat...\n";
    });
}

数据库操作

public function onWorkerStart($worker)
{
    // 初始化数据库连接
    \think\facade\Db::connect();
}
public function onMessage(TcpConnection $connection, $data)
{
    // 数据库操作
    $result = Db::table('users')->where('id', 1)->find();
    $connection->send(json_encode($result));
}