完整版websocket带数据库

swoolechat.php

<?php
use SwooleCoroutine as co;
class swoolechat {
    //协程内部的变量
    public $sql_string=null;
    public $sql_res=null;
    public $map=null;
    //公共变量
    public $ws;
    public $configws=array(
        'worker_num' => 1,//进程数 进程数越多,负载能力越大,建议cpu核数2倍
        'max_request' => 10000,//进程处理完这么多任务数,就会重启,防止溢出
        'max_conn'=>10000,//最大保持连接数,也就是同时在线人数
        'ipc_mode' => 1,//1使用unix socket通信2使用消息队列通信3使用消息队列通信，并设置为争抢模式
        'dispatch_mode' => 2,//2固定模式,同一个链接只被一个worker处理,1轮询模式3争抢模式
        'debug_mode'=> 1 ,
        'daemonize' => 0,//守护模式
        'log_file' => 'logswoole.log',
        'heartbeat_check_interval' => 30,//心跳侦测时间间隔
        'heartbeat_idle_time' => 600,//闲置时间上线
    );
    public $configsql=array(
        'host'=>'127.0.0.1',
        'port'=>3306,
        'user'=>'ws_ltcpi_com',
        'password'=>'La2bKpT4T2xrrkLL',
        'database'=>'ws_ltcpi_com',
        'charset'=>'utf8',
        'timeout'=>10,
    );
    //构造函数
    public function __construct() {
        //创建websocket服务器对象，监听0.0.0.0:端口
        $this->ws = new swoole_websocket_server("0.0.0.0", 9501);
        $this->ws->set($this->configws);
        //监听WebSocket连接
        $this->ws->on('open', function ($ws, $request) {
            echo "server: handshake success with fd{$request->fd}
";
            $this->ws->push($request->fd, "hello, welcome
");
        });
        //监听WebSocket收到信息
        $this->ws->on('message', function ($ws, $frame) {
//            echo "onmessage: handshake success with fd{$frame}
";
            self::onmessage($ws,$frame->fd,$frame->data);
        });
        //关闭事件
        $this->ws->on('close', function ($ws, $fd) {
            echo "client-{$fd} is closed
";
        });
        //启动服务
        echo "启动3服务
";
        $this->ws->start();
    }
    //内部方法
    protected function get_sql_insert($table,$arr){
        $sql=" insert into ".$table." (";
        foreach (array_keys($arr) as $k=>$v){
            $sql.=$v.',';
        }
        $sql=rtrim($sql, ",");
        $sql.=') values (';
        foreach ($arr as $k=>$v){
            $sql.="'".$v."',";
        }
        $sql=rtrim($sql, ",");
        $sql.=')';
        return $sql;
    }
    protected function get_sql_delete_fd($fd){
        $res="DELETE FROM fa_fdband WHERE fd = '{$fd}'";
        return $res;
    }
    protected function get_sql_delete_userid($userid){
        $res="DELETE FROM fa_fdband WHERE userid = '{$userid}'";
        return $res;
    }
    protected function pushtofd($fd,$type,$status,$msg){
        $res=json_encode(array('type'=>$type,'status'=>$status,'msg'=>$msg));
        $this->ws->push($fd,utf8_encode($res));
    }
    protected function get_sql_bandlist($fd){
        $sql="select * from fa_fdband where fd='{$fd}'order by id desc";
        return $sql;
    }
    //收到命令
    public function onmessage($ws,$fd,$data){
        //获取fd  数组数据
        $data_arr=json_decode($data, true);
        if ($data=='heart' or array_key_exists('type',$data_arr)==false){
            //心跳回复
            $ws->push($fd,'heart');
        }else{
            $type=$data_arr['type'];
            //登录房间
            if ($type=='band'){
                $roomid=$data_arr['roomid'];
                $userid=$data_arr['userid'];
                self::band($fd,$roomid,$userid);
            }
            //聊天
            if ($type=='chatsend'){
                $msg=$data_arr['msg'];
                self::chatsend($fd,$msg);
            }
        }
    }
    //绑定房间
    public function band($fd,$roomid,$userid){
        //生成删除语句  插入语句
        $this->map=array(
            'fd'=>$fd,
            'userid'=>$userid,
            'roomid'=>$roomid,
            'createtime'=>time(),
        );
        go(function() {
            $db=new coMySQL();
            $db->connect($this->configsql);
            //获取变量
            $fd=$this->map['fd'];
            $roomid=$this->map['roomid'];
            $userid=$this->map['userid'];
            //删除同fd房间
            $res=$db->query(self::get_sql_delete_fd($fd));
            var_dump(self::get_sql_delete_fd($fd));
            var_dump($res);
            if ($res!=1){
                self::pushtofd($fd,'band',0,'登录房间失败,删除同fd失败');
            }
            //删除同userid绑定信息
            $res=$db->query(self::get_sql_delete_userid($userid));
            if ($res!=1){
                self::pushtofd($fd,'band',0,'登录房间失败,删除同userid失败');
            }
            //创建房间
            $res=$db->query(self::get_sql_insert('fa_fdband',$this->map));
            if ($res==1){
                self::pushtofd($this->map['fd'],'band',1,'登录房间成功');
            }else{
                self::pushtofd($this->map['fd'],'band',0,'登录房间失败,因为插入失败');
            }
        });
    }

    public function chatsend($fd,$msg){
        $this->map=array(
            'fd'=>$fd,
            'msg'=>$msg,
        );
        go(function() {
            $db=new coMySQL();
            $db->connect($this->configsql);
            //获取变量
            $fd=$this->map['fd'];
            $msg=$this->map['msg'];
            //根据fd查询绑定信息
            $band_list=$db->query(self::get_sql_bandlist($fd));
            if (count($band_list)<1){
                self::pushtofd($this->map['fd'],'chatsend',0,'请先绑定房间');
                return;
            }
            //保存记录
            $map=array(
                'userid'=>$band_list[0]['userid'],
                'roomid'=>$band_list[0]['roomid'],
                'msg'=>$msg,
                'createtime'=>time(),
            );
            //生成保存消息语3句
            $res=$db->query(self::get_sql_insert('fa_fdchat',$map));
            var_dump(self::get_sql_insert('fa_fdchat',$map));
            var_dump($res);
            self::pushtofd($this->map['fd'],'chatsend',1,'success');
            //转发消息
            $band_list=$db->query("select * from fa_fdband where roomid='{$map['roomid']}'order by id desc");
            foreach ($band_list as $k=>$v){
                if ($v['fd']==$fd){
                    $msg=array(
                        'nickname'=>'我',
                        'msg'=>$map['msg']
                    );
                    self::pushtofd($v['fd'],'chatrev',1,$msg);
                }else{
                    $msg=array(
                        'nickname'=>'追风少年',
                        'msg'=>$map['msg']
                    );
                    self::pushtofd($v['fd'],'chatrev',1,$msg);
                }
            }
        });
    }
}
$obj=new SwooleChat();

client.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>webSocket客户端接收命令</title>
</head>
<body>
<p id="status"></p>
<input type="text" id="login" style=" 500px"  value='{"type":"band","roomid":"3","userid":"4"}'/>
<button onclick="login()">绑定房间</button>
<br/>
<input type="text" id="chat" style=" 500px"  value='{"type":"chatsend","msg":"大家好我是新人4"}'/>
<button onclick="chat()">发送消息</button>
<!--<button onclick="clientClose()">关闭连接</button>-->
<!--<button onclick="clientBreak()">彻底断开</button>-->
<script type="text/javascript">
    var lockReconnect = false;//避免重复连接
    var wsUrl = "ws://124.156.153.202:9501";		// websocket链接
    var ws;
    function decodeUnicode(str) {
        str = str.replace(/\/g, "%");
        return unescape(str);
    }

    createWebSocket();
    // 创建websocket
    function createWebSocket(){
        try {
            ws = new WebSocket(wsUrl);
            ws.onopen = function (evt) {
                onOpen(evt);
            };
            ws.onclose = function (evt) {
                onClose(evt);
                websocketReconnect(wsUrl);
            };
            // 接收服务端数据时触发事件
            ws.onmessage = function (evt) {
                onMessage(evt);
            };
            // 通信发生错误时触发
            ws.onerror = function (evt) {
                websocketReconnect(wsUrl);
                onError(evt);
            };
        } catch (e) {
            console.log('catch');
            websocketReconnect(wsUrl);
        }
    }
    //被动业务
    function onOpen(evt) {
        document.getElementById('status').innerText += "receive：" + "建立链接成功..." + "
";
        //心跳检测重置
        heartCheck.start();
    }
    function onClose(evt) {
        document.getElementById('status').innerText += "receive：" + "连接已关闭..." + "
";
    }
    function onMessage(evt) {
        if (evt.data!=='heart'){

            document.getElementById('status').innerText += "receive：" + decodeUnicode(evt.data)+ "
";
        }
        heartCheck.start();
    }
    function onError(evt) {
        document.getElementById('status').innerText += "通信错误"+ "
";
    }
    //重新连接
    function websocketReconnect(url) {
        if (lockReconnect) {       // 是否已经执行重连
            return;
        }else {
            lockReconnect = true;
            //没连接上会一直重连，设置延迟避免请求过多
            tt && clearTimeout(tt);
            var tt = setTimeout(function () {
                createWebSocket(url);
                lockReconnect = false;
            }, 5000);
        }
    }
    //发送心跳
    var heartCheck = {
        timeout: 5000,
        timeoutObj: null,
        serverTimeoutObj: null,
        start: function () {
            var self = this;
            this.timeoutObj && clearTimeout(this.timeoutObj);
            this.serverTimeoutObj && clearTimeout(this.serverTimeoutObj);
            this.timeoutObj = setTimeout(function () {
                ws.send("heart");
                console.log('send:start');
                self.serverTimeoutObj = setTimeout(function () {
                    ws.close();
                }, self.timeout);
            }, this.timeout)
        }
    };
    //主动业务
    function login() {
        ws.send(document.getElementById('login').value);
    }
    function chat() {
        ws.send(document.getElementById('chat').value);
    }

    function clientClose() {
        ws.close();
        document.getElementById('status').innerText += "客户端主动普通断开" +"
";
    }

</script>
</body>
</html>

sql

https://files.cnblogs.com/files/bufeetu/websocket.7z