laravel整合workerman做聊天室

测试工具  http://www.blue-zero.com/WebSocket/

2018年8月6日17:28:24

<?php

namespace AppConsoleCommands;

use IlluminateConsoleCommand;
use WorkermanWorker;
use AppWorkChatroomWork;

class Chatroom extends Command {

    protected $taskserver;
    /*
     * 操作参数
     * 注意只能在
     * start 启动
     * stop 停止
     * relaod  只能重启逻辑代码，核心workerman_init无法重启，注意看官方文档
     * status 查看状态
     * connections 查看连接状态（需要Workerman版本>=3.5.0）
     * 
     * 库 composer require workerman/workerman
     */
    protected $action = array('start', 'stop', 'reload', 'status', 'connections');

    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'Chatroom {action}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = 'Chatroom';

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct() {
        parent::__construct();
    }

    /**
     * Execute the console command.
     *
     * @return mixed
     * 
     * 注意上线提供的方法
     * 
     */
    public function handle() {
        $action = $this->argument('action');

        if (!in_array($action, $this->action)) {
            $this->error('Error Action');
            exit;
        }
        //初始化workerman
        ChatroomWork::workerman_init($action);
    }

}

<?php

namespace AppWork;

use AppWorkBaseWork as Base;
use IlluminateSupportFacadesDB;
use AppWorkCommonWork;
use WorkermanWorker;
use WorkermanLibTimer;
use AppModelsOperationLog;
use AppModelsUsers;

class ChatroomWork extends Base {

    //全局的总连接数
    static $connection_count = 0;
    //单个房间最大的连接数
    static $room_max_numbers;
    //房间所有的用户连接数ID集合，判断发送给那些用户，绑定用户和connection_id
    static $room_connection_array = [];

    public static function workerman_init($action = null) {
        global $argv;

        $argv[0] = 'workerman:websocket';
        $argv[1] = $action;
//        $argv[2] = '-d';
//        心跳
        define('HEARTBEAT_TIME', 30);
        //初始化
        $worker = new Worker("websocket://172.17.1.247:9090");
        $worker->name = 'Chatroom';
        //linux 用户线上是www
//        $worker->user = 'www';
        //守护模式信息输出文件地址
//        $worker->stdoutFile = "./workerman.log";
        //工作进程总数 测试环境4个
        $worker->count = 4;
        //正式环境
//        $ws->count = 10;
        //建立链接 处理逻辑
        $worker->onConnect = function($connection) {
//             有新的客户端连接时，连接数+1
            self::$connection_count++;
            self::onConnect($connection);
        };
        //接受消息 处理逻辑
        $worker->onMessage = function($connection, $data) {

            self::onMessage($connection, $data);
        };
        //关闭链接 处理逻辑
        $worker->onClose = function($connection) {
//            客户端关闭时，连接数-1
            self::$connection_count--;
            self::onClose($connection);
        };

        // 进程启动后设置一个30秒运行一次的定时器
        $worker->onWorkerStart = function($worker) {
//            Timer::add(1, function()use($worker) {
//                $time_now = time();
//                foreach ($worker->connections as $connection) {
//                 
//                }
//            });
        };
        // 开始
        Worker::runAll();
    }

    //建立链接 处理逻辑
    public static function onConnect($connection) {
        //主动心跳ping测试60秒一次
//        Timer::add(HEARTBEAT_TIME, function() use($connection) {
//            $connection->send(json_encode(['code' => 200, 'msg' => '服务存活', 'data' => [], 'connections' => self::$connection_count]));
//        });
    }

    //接受消息 处理逻辑
    public static function onMessage($connection, $data) {
        //解析数据，非合法的json数据不处理

        if (!empty($data)) {
            if (is_json($data)) {
                $data = json_decode($data, true);

                if ($data['action_type'] == 'ping') {
                    // 客户端回应服务端的心跳
                    $connection->send(json_encode(['code' => 200, 'msg' => '服务存活', 'data' => [], 'connections' => self::$connection_count]));
                } elseif ($data['action_type'] == 'login') {
                    //用户登录
                    if ($data['is_login'] == 1) {
                        //匿名登录
                        self::$room_connection_array[$data['room_id']][$connection->id]['user_name'] = '匿名用户' . $connection->id;

                        $connection->send(json_encode(['code' => 200, 'msg' => '匿名登录成功', 'data' => self::$room_connection_array, 'connections' => self::$connection_count]));
                    } elseif ($data['is_login'] == 2) {
                        //已登录
                        $Users = Users::where('id', $data['user_id'])->first();
                        if (empty($Users)) {
                            $connection->send(json_encode(['code' => 201, 'msg' => '用户ID无效或者错误', 'data' => [], 'connections' => self::$connection_count]));
                        } else {
                            $Users = $Users->toArray();
                            self::$room_connection_array[$data['room_id']][$connection->id]['user_name'] = empty($Users['realname']) ? $data['user_id'] : $Users['realname'];

                            $connection->send(json_encode(['code' => 200, 'msg' => '登录成功', 'data' => self::$room_connection_array, 'connections' => self::$connection_count]));
                        }
                    } else {
                        $connection->send(json_encode(['code' => 201, 'msg' => '登录类型数据错误', 'data' => [], 'connections' => self::$connection_count]));
                    }
                } elseif ($data['action_type'] == 'broadcast_to_all') {
                    //只发给房间的所有的人，除去自己
                    foreach ($connection->worker->connections as $con) {
                        foreach (self::$room_connection_array[$data['room_id']] as $k => $v) {
//                            p($con->id);
//                            p($k);
                            if ($k == $con->id && $con->id != $connection->id) {
                                $con->send($data['message']);
                            }
                        }
                    }
                } elseif ($data['action_type'] == 'broadcast_to_one') {
                    
                } else {
                    $connection->send(json_encode(['code' => 201, 'msg' => 'action_type类型错误', 'data' => [], 'connections' => self::$connection_count]));
                }
            }
        } else {
            $connection->send(json_encode(['code' => 201, 'msg' => '数据请求为空', 'data' => [], 'connections' => self::$connection_count]));
        }
    }

    //关闭链接 处理逻辑
    public static function onClose($connection) {
        //检索$connection->id 是否有在self::$room_connection_array中，有的话就剔除

        foreach (self::$room_connection_array as $k => $v) {
            foreach ($v as $kk => $vv) {
                if ($kk == $connection->id) {
                    unset(self::$room_connection_array[$k][$connection->id]);
                }
            }
        }
//        p(self::$room_connection_array['100']);
    }

}

{"action_type":"login","is_login":1,"room_id":101}  //登录


{"action_type":"broadcast_to_all","room_id":100,"message":"111"} //发送消息

为什么不用switch代码会看起来更清晰，因为有bug，对字符串匹配的不好

注意：
1，$connection就是当前的连接数据
2，因为未根据work id做房间划分，不知道在超出单个work连接时候会不会出问题
3，可以根据实际压力去划分一个work的最大连接数，这里是简单的测试demo所以未做具体的细节划分
4，这里应该结合session来做处理数据，但是我只是根据发送数据来区别用户，你可以在登录的发送数据的时候根据session处理数据，看需要，必须要返回用户列表，就直接把room id下面的所有用户名返回就OK
5，workerman用起来其实还是比较简单的，但是我这种结合laravel的整合是有问题，比如现在我有一个消息推送，一个聊天室就没办法放在一起使用，必须用别的办法，如果是单功能就比较容易，直接结合，我有一个骚办法就是直接
复制artisan入口文件，直接增加新入口artisan1，经过测试完全没有问题，但是其实不是很好的解决方案，如果要用就先这么上吧

如果你觉得麻烦可以是gateway做比较简单

2019年7月12日09:43:56

 注意：上面是临时测试代码。业务代码使用try catch处理异常和错误