swoole 多進程共享數據

 

進程作為程序執行過程中資源分配的基本單位,擁有獨立的地址空間,同一進程的線程可以共享本進程的全局變量,靜態變量等數據和地址空間,但進程之間資源相互獨立.由於PHP語言不支持多線程,因此Swoole使用多進程模式,再多進程模式下就存在進程內存隔離,進程間通信與數據共享問題.

swoole中master主進程會創建manager管理進程和reactor線程,真正的工作進程為worker進程.  manager是創建和管理worker進程,reactor進程測試監聽socket,接受數據任務,發送給worker進程去工作,因此所有業務邏輯最終都是在worker進程中進行的,worker進程之間的數據共享與通信必不可少.

 swoole中 設置選項worker_num設置 啟動的worker進程數,默認設置為CPU核數

$server = new swoole_server('127.0.0.1',9898);
$server->set(array(
    'worker_num' => 4,   //設置啟動的Worker進程數。
));

如上面說描述,進程存在進程隔離:

$fds = array();
$server->on('connect', function ($server, $fd){
    echo "connection open: {$fd}\n";
    global $fds;
    $fds[] = $fd;
    var_dump($fds);
});

$fds雖然是全局變量,但是只在但前的進程內有效,swoole服務器底層會創建多個worker進程,此處打印出來的只有部分連接的fd,本文講簡述兩種解決方案的簡單示例:

1.外部存儲服務 : Redis

作為內存數據庫redis 無太多IO等待,並且讀寫速度快

示例代碼:以簡易聊天室websocket服務 swoole_websocket_server為例

 1 $ws = new swoole_websocket_server("0.0.0.0", 9999);
$redis = new \Redis();
$redis->connect('127.0.0.1', 6379);
$ws->set(array(
    'daemonize' => false,
    'worker_num' => 4,
));
//監聽WebSocket連接打開事件
$ws->on('open', function ($ws, $request) use($redis) {
    var_dump($request->fd, $request->get, $request->server);
    //記錄連接
    $redis->sAdd('fd', $request->fd);
    $count = $redis->sCard('fd');
    var_dump($count);
    $ws->push($request->fd, 'hello, welcome ☺                     當前'.$count.'人連接在線');
});
//監聽WebSocket消息事件
$ws->on('message', function ($ws, $frame) use($redis) {
    $fds  = $redis->sMembers('fd');
    $data = json_decode($frame->data,true);
    if($data['type'] ==1 ){
        $redis->set($frame->fd,json_encode(['fd'=>$frame->fd,'user'=>$data['user']]));
        //通知所有用戶新用戶上線
        $fds = $redis->sMembers('fd');$users=[];
        $i=0;
        foreach ($fds as $fd_on){
            $info = $redis->get($fd_on);
            $users[$i]['fd']   = $fd_on;
            $users[$i]['name'] = json_decode($info,true)['user'];
            $message = "歡迎 <b style='color: darkmagenta ;'>".$data['user']."</b> 進入聊天室";
            $push_data = ['message'=>$message,'users'=>$users];
            $ws->push($fd_on,json_encode($push_data));
            $i++;
        }
    }else if($data['type'] ==2){
        if($data['to_user'] == 'all'){
            foreach ($fds as $fd){
                $message = "<b style='color: crimson'>".$data['from_user']." say:</b>  ".$data['msg'];
                $push_data = ['message'=>$message];
                $ws->push($fd,json_encode($push_data));
            }
        }
    }
    echo "Message: {$frame->data}\n";
});
//監聽WebSocket連接關閉事件
$ws->on('close', function ($ws, $fd) use ($redis){
    $redis->sRem('fd',$fd);
    $fds = $redis->sMembers('fd');
    $i=0;
    foreach ($fds as $fd_on){
        $user = json_decode($redis->get($fd),true)['user'];
        $info = $redis->get($fd_on);
        $users[$i]['fd']   = $fd_on;
        $users[$i]['name'] = json_decode($info,true)['user'];
        $message = "<b style='color: blueviolet'>".$user."</b> 離開聊天室了";
        $push_data = ['message'=>$message,'users'=>$users];
        $ws->push($fd_on,json_encode($push_data));
        $i++;
    }
    echo "client-{$fd} is closed\n";
});

 

2.共享內存拓展:swoole_table

swoole_table是swoole官方提供的基於共享內存和鎖實現的超高性能冰飯數據結構.swoole_table在swoole1.7.5版本后可用.

目前swoole只支持3種類型:

swoole_table::TYPE_INT 整形字段

swoole_table::TYPE_FLOAT浮點字段

swoole_table::TYPE_STRING 字符串字段

 

函數方法:

column() :給內存表增加一列 參數:字段名,字段類型,字節數

$table->column('id', swoole_table::TYPE_INT, 4);

create():基於前一步對表結構的創建,執行創建表.

set() :設置行的數據(key-value的方式) 參數: 數據的key,數據的值(必須數組,鍵名必須與字段定義的$name相同)

$table->set($fd, ['id'=>1]);

get() :獲取一行數據  參數:數據的key

$table->get($fd);

del() :刪除一行數據 參數:數據的key

$table->del($fd);

lock():鎖定整個表

unlock():釋放鎖

lock/unlock 必須成對出現,否則會發生死鎖.

示例代碼: 還是上面的websocket服務為例

class WebSocketServer {
    private $server;
    public function __construct()
    {
        $this->server = new swoole_websocket_server("0.0.0.0",9988);
        $this->server->set(array(
            'daemonize'       => false,
            'worker_num'      => 4,
        ));
        //內存表
        $fd_table = new swoole_table( 1024 );
        $fd_table->column( "user",swoole_table::TYPE_STRING, 30 );
        $fd_table->column( "time", swoole_table::TYPE_STRING, 20 );
        $fd_table->create();

        $user_table = new swoole_table(1024);
        $user_table->column("fd",swoole_table::TYPE_INT,8);
        $user_table->create();

        $this->server->fd = $fd_table;
        $this->server->user = $user_table;

        //啟動開始
        $this->server->on('Start',[$this,'onStart']);
        //與onStart同級
        $this->server->on('workerStart',[$this,'onWorkerStart']);
        //webSocket open 連接觸發回調
        $this->server->on('open',[$this,'onOpen']);
        //webSocket send 發送觸發回調
        $this->server->on('message', [$this, 'onMessage']);
        //webSocket close 關閉觸發回調
        $this->server->on('Close', [$this, 'onClose']);
        //tcp連接 觸發 在 webSocket open 之前回調
        $this->server->on('Connect', [$this, 'onConnect']);
        //tcp 模式下（eg:telnet ） 發送信息才會觸發  webSocket 模式下沒有觸發
        $this->server->on('Receive', [$this, 'onReceive']);
        // 服務開啟
        $this->server->start();

    }

    public function onStart( $server)
    {
        echo "Start\n";
    }

    public function onWorkerStart($server,$worker_id)
    {
        //判斷是worker進程還是 task_worker進程 echo 次數 是worker_num+task_worker_num
        if($worker_id<$server->setting['worker_num']){
            echo  'worder'.$worker_id."\n";
        }else{
            echo  'task_worker'.$worker_id."\n";
        }
        //     echo "workerStart{$worker_id}\n";
    }

    public function onOpen( $server,$request)
    {
        $this->server->fd->set($request->fd,['user'=>'']);
        echo "server: handshake success with fd{$request->fd}\n";
        $count = count($server->connections);
        $server->push($request->fd, 'hello, welcome ☺                     當前'.$count.'人連接在線');
    }

    public function onMessage( $server,$frame)
    {
        echo "receive from {$frame->fd}:{$frame->data},opcode:{$frame->opcode},fin:{$frame->finish}\n";
        $data = json_decode($frame->data,true);
        if($data['type'] ==1 ){
            $server->fd->set($frame->fd,['user'=>$data['user']]);
            //通知所有用戶新用戶上線
            foreach($server->connections as $key => $fd) {
                $server->push($fd, "歡迎 <b style='color: darkmagenta ;'>".$data['user']."</b> 進入聊天室");
            }
        }else if($data['type'] ==2){
            if($data['to_user'] == 'all'){
                foreach($server->connections as $key => $fd) {
                    $server->push($fd, "<b style='color: crimson'>".$data['from_user']." say:</b>  ".$data['msg']);
                }
            }
        }
    }


    public function onConnect( $server, $fd, $from_id ) {
        echo "Client {$fd} connect\n";
        echo "{$from_id}\n";
    }

    public function onReceive( $server, $fd, $from_id, $data ) {
        echo "Get Message From Client {$fd}:{$data}\n";
    }


    public function onClose($server, $fd)
    {
        echo "Client {$fd} close connection\n";
        foreach($server->connections as $key => $on_fd) {
            $user = $server->fd->get($fd)['user'];
            $server->push($on_fd, "<b style='color: blueviolet'>".$user."</b> 離開聊天室了");
        }
    }
}
new WebSocketServer();