基于swoole的网页一对一实时聊天

CREATE TABLE `msg` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `content` varchar(255) NOT NULL DEFAULT ‘‘ COMMENT ‘内容‘,
  `tid` int(11) NOT NULL DEFAULT ‘0‘ COMMENT ‘接收用户id‘,
  `fid` int(11) NOT NULL DEFAULT ‘0‘ COMMENT ‘发送用户id‘,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=29 DEFAULT CHARSET=utf8 COMMENT=‘消息表‘;

CREATE TABLE `fd` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `uid` int(11) NOT NULL DEFAULT ‘0‘ COMMENT ‘用户id‘,
  `fd` int(11) NOT NULL DEFAULT ‘0‘ COMMENT ‘绑定id‘,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8 COMMENT=‘用户绑定表‘;

initDb();
        $this->serv = new swoole_websocket_server("0.0.0.0", 9502);
        $this->serv->set(array(
            ‘worker_num‘ => 8,
            ‘daemonize‘ => false,
            ‘max_request‘ => 10000,
            ‘dispatch_mode‘ => 2,
            ‘debug_mode‘ => 1
        ));

        $this->serv->on(‘Open‘, array($this, ‘onOpen‘));
        $this->serv->on(‘Message‘, array($this, ‘onMessage‘));
        $this->serv->on(‘Close‘, array($this, ‘onClose‘));

        $this->serv->start();

    }

    function onOpen($server, $req)
    {
        // $server->push($req->fd, json_encode(33));
    }

    public function onMessage($server, $frame)
    {
        //$server->push($frame->fd, json_encode(["hello", "world"]));
        $pData = json_decode($frame->data);
        $data = array();
        if (isset($pData->content)) {
            $tfd = $this->getFd($pData->tid); //获取绑定的fd
            $data = $this->add($pData->fid, $pData->tid, $pData->content); //保存消息
            $server->push($tfd, json_encode($data)); //推送到接收者
        } else {
            $this->unBind(null,$pData->fid); //首次接入，清除绑定数据
            if ($this->bind($pData->fid, $frame->fd)) {  //绑定fd
                $data = $this->loadHistory($pData->fid, $pData->tid); //加载历史记录
            } else {
                $data = array("content" => "无法绑定fd");
            }
        }
        $server->push($frame->fd, json_encode($data)); //推送到发送者

    }


    public function onClose($server, $fd)
    {
        $this->unBind($fd);
        echo "connection close: " . $fd;
    }


    /*******************/
    function initDb()
    {
        $conn = mysqli_connect("192.168.1.122", "root", "a123456");
        if (!$conn) {
            die(‘Could not connect: ‘ . mysql_error());
        } else {
            mysqli_select_db($conn, "test");
        }
        $this->conn = $conn;
    }

    public function add($fid, $tid, $content)
    {
        $sql = "insert into msg (fid,tid,content) values ($fid,$tid,‘$content‘)";
        if ($this->conn->query($sql)) {
            $id = $this->conn->insert_id;
            $data = $this->loadHistory($fid, $tid, $id);
            return $data;
        }
    }

    public function bind($uid, $fd)
    {
        $sql = "insert into fd (uid,fd) values ($uid,$fd)";
        if ($this->conn->query($sql)) {
            return true;
        }
    }

    public function getFd($uid)
    {
        $sql = "select * from fd where uid=$uid limit 1";
        $row = "";
        if ($query = $this->conn->query($sql)) {
            $data = mysqli_fetch_assoc($query);
            $row = $data[‘fd‘];
        }
        return $row;
    }

    public function unBind($fd, $uid = null)
    {
        if ($uid) {
            $sql = "delete from fd where uid=$uid";
        } else {
            $sql = "delete from fd where fd=$fd";
        }
        if ($this->conn->query($sql)) {
            return true;
        }
    }

    public function loadHistory($fid, $tid, $id = null)
    {
        $and = $id ? " and id=$id" : ‘‘;
        $sql = "select * from msg where ((fid=$fid and tid = $tid) or (tid=$fid and fid = $tid))" . $and;
        $data = [];
        if ($query = $this->conn->query($sql)) {
            while ($row = mysqli_fetch_assoc($query)) {
                $data[] = $row;
            }
        }
        return $data;
    }
}

// 启动服务器
$server = new Server();

php Server.php

发送

if($server->push($tfd, json_encode($data))){
      //标记已读  
}