基於swoole的網頁一對一實時聊天

需求分析

網站上實現一對一即時溝通，能查看聊天記錄以及離線留言，新消息提醒。

核心技術

html5的websocket，php的swoole擴展http://wiki.swoole.com/

數據表

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='用戶綁定表';

Server端代碼

<?php

class Server
{
    private $serv;
    private $conn = null;
    private static $fd = null;

    public function __construct()
    {
        $this->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();

備注：swoole_websocket_server是基於tcp的長連接，僅支持cli模式運行。

啟動服務器

php Server.php

客戶端代碼

<!DOCTYPE html>
<html>
<head>
    <title></title>
    <meta charset="UTF-8">
    <script src="jquery-2.1.1.min.js"></script>
    <script src="jquery.json.js"></script> /*js地址：https://files.cnblogs.com/files/zenghansen/jquery.json-2.3.min.js*/

    <script type="text/javascript">
        var fid = 1; //發送者uid var tid = 2; //接收者uid var exampleSocket = new WebSocket("ws://192.168.1.17:9502"); $(function () { exampleSocket.onopen = function (event) { console.log(event.data); initData(); //加載歷史記錄 }; exampleSocket.onmessage = function (event) { console.log(event.data); loadData($.parseJSON(event.data)); //導入消息記錄，加載新的消息 } }) function sendMsg() { var pData = { content: document.getElementById('content').value, fid: fid, tid: tid, } if(pData.content == ''){ alert("消息不能為空"); return; } exampleSocket.send($.toJSON(pData)); //發送消息 } function initData() { var pData = { fid: fid, tid: tid, } exampleSocket.send($.toJSON(pData)); //獲取消息記錄，綁定fd } function loadData(data) { for (var i = 0; i < data.length; i++) { var html = '<p>' + data[i].fid + '>' + data[i].tid + ':' + data[i].content + '</p>'; $("#history").append(html); } } </script>
</head>
<body>
<div id="history" style="border: 1px solid #ccc; width: 100px; height: auto">

</div>
<input type="text" id="content">
<button onclick="sendMsg()">發送</button>
</body>
</html>

 

ps1：再復制一份客戶端，修改一下發送者你接收者的uid，即可進行模擬實時聊天。

ps2：此代碼已經實現了加載歷史記錄的功能

ps3：若要增加新消息提醒功能，msg還需增加一個已讀標示，然后推送給接收者的時候　

if($server->push($tfd, json_encode($data))){
      //標記已讀  
}    

ps4：然后沒有標記已讀的消息，就是新消息提醒。

如有疑問請加作者qq：375161864