laravel整合workerman做聊天室

測試工具  http://www.blue-zero.com/WebSocket/

2018年8月6日17:28:24

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Workerman\Worker;
use App\Work\ChatroomWork;

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 App\Work;

use App\Work\BaseWork as Base;
use Illuminate\Support\Facades\DB;
use App\Work\CommonWork;
use Workerman\Worker;
use Workerman\Lib\Timer;
use App\Models\OperationLog;
use App\Models\Users;

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處理異常和錯誤