摘要：目前網(wǎng)站有兩個(gè)用到實(shí)時(shí)消息推送的功能，源碼最新動(dòng)態(tài)，實(shí)時(shí)顯示用戶的操作行為消息推送，如重要消息通知，任務(wù)指派等等考慮的問題既然要實(shí)現(xiàn)即時(shí)，那就少不了。

目前網(wǎng)站有兩個(gè)用到實(shí)時(shí)消息推送的功能，源碼：https://github.com/wuzhc/team

最新動(dòng)態(tài)，實(shí)時(shí)顯示用戶的操作行為

消息推送，如重要消息通知，任務(wù)指派等等

既然要實(shí)現(xiàn)即時(shí)，那就少不了socketio。因?yàn)轫?xiàng)目是PHP寫的，所以服務(wù)端直接用phpsocket.io

我們應(yīng)該保存離線消息，否則如果用戶不在線，那就接受不到消息。這里我用mongodb來存儲(chǔ)消息。

一是消息不需要關(guān)聯(lián)表，一條消息一個(gè)文檔

二是mongodb適合做海量數(shù)據(jù)存儲(chǔ)，并且分片也很簡單

三是消息不需要永久存儲(chǔ)，隨著時(shí)間推移，消息的價(jià)值性越低，可以用固定集合來存儲(chǔ)消息，當(dāng)數(shù)據(jù)量達(dá)到一定值時(shí)覆蓋最久的消息

一個(gè)頁面連接一個(gè)socket，若用戶同時(shí)打開了多個(gè)頁面（即一個(gè)用戶會(huì)對應(yīng)多個(gè)socketID），那么消息應(yīng)該推送到用戶每一個(gè)打開的頁面。一個(gè)簡單的做法就是把用戶多個(gè)sockID加到group分組（socket.emit("group")），group分組名稱可以是"uid:1"，表示userID為1的用戶，然后socket.broadcast.to("uid:1").emit("event_name", data);

就可以實(shí)現(xiàn)對多個(gè)頁面推送消息

        // 初始化io對象
        var socket = io("http://" + document.domain + ":2120");
        // 當(dāng)socket連接后發(fā)送登錄請求
        socket.on("connect", function () {
            socket.emit("login", {
                uid: "user->id?>",
                companyID: "user->identity->fdCompanyID?>"
            });
        });
        // 實(shí)時(shí)消息，當(dāng)服務(wù)端推送來消息時(shí)觸發(fā)
        socket.on("new_msg", function (msg) {
            if (msg.typeID == "") {
                var html = "
" +
                    "" +
                    "
" +
                    "" +
                    "
" +
                    "
" + msg.title + "
" +
                    "
" + msg.content + "
" +
                    "" +
                    "
";
                $("#msg-handle").prepend(html);

                var num = $(".msg-handle-num").eq(1).text() * 1;
                $(".msg-handle-num").text(num + 1);
            }
        });
        
        // 實(shí)時(shí)動(dòng)態(tài)，當(dāng)服務(wù)端推送來消息時(shí)觸發(fā)
        socket.on("update_dynamic", function (msg) {
            var html = "
" +
                "" +
                "" +
                "
" +
                "" +
                "" + msg.date + "" +
                "
" +
                "" + msg.operator + "" +
                "" + msg.title + "" +
                "
" +
                "
" + msg.content + "
" +
                "
" +
                "
";
            $("#dynamic-list").prepend(html);
        });
        
        // 在線人數(shù)
        socket.on("update_online_count", function (msg) {
            $("#online-people").html(msg);
        });

    /**
     * 消息推送
     */
    public function actionStart()
    {
        // PHPSocketIO服務(wù)
        $io = new SocketIO(2120);
        // 客戶端發(fā)起連接事件時(shí)，設(shè)置連接socket的各種事件回調(diào)
        $io->on("connection", function ($socket) use ($io) {

            /** @var Connection $redis */
            $redis = Yii::$app->redis;

            // 當(dāng)客戶端發(fā)來登錄事件時(shí)觸發(fā)
            $socket->on("login", function ($data) use ($socket, $redis, $io) {

                $uid = isset($data["uid"]) ? $data["uid"] : null;
                $companyID = isset($data["companyID"]) ? $data["companyID"] : null;

                // uid和companyID應(yīng)該做下加密 by wuzhc 2018-01-28
                if (empty($uid) || empty($companyID)) {
                    return ;
                }
                $this->companyID = $companyID;

                // 已經(jīng)登錄過了
                if (isset($socket->uid)) {
                    return;
                }

                $key = "socketio:company:" . $companyID;
                $field = "uid:" . $uid;

                // 用hash方便統(tǒng)計(jì)用戶打開頁面數(shù)量
                if (!$redis->hget($key, $field)) {
                    $redis->hset($key, $field, 0);
                }

                // 同個(gè)用戶打開新頁面時(shí)加1
                $redis->hincrby($key, $field, 1);

                // 加入uid分組，方便對同個(gè)用戶的所有打開頁面推送消息
                $socket->join("uid:". $uid);

                // 加入companyID，方便對整個(gè)公司的所有用戶推送消息
                $socket->join("company:".$companyID);

                $socket->uid = $uid;
                $socket->companyID = $companyID;

                // 整個(gè)公司在線人數(shù)更新
                $io->to("company:".$companyID)->emit("update_online_count", $redis->hlen($key));
            });

            // 當(dāng)客戶端斷開連接是觸發(fā)（一般是關(guān)閉網(wǎng)頁或者跳轉(zhuǎn)刷新導(dǎo)致）
            $socket->on("disconnect", function () use ($socket, $redis, $io) {
                if (!isset($socket->uid)) {
                    return;
                }

                $key = "socketio:company:" . $socket->companyID;
                $field = "uid:" . $socket->uid;
                $redis->hincrby($key, $field, -1);

                if ($redis->hget($key, $field) <= 0) {
                    $redis->hdel($key, $field);

                    // 某某下線了，刷新整個(gè)公司的在線人數(shù)
                    $io->to("company:".$socket->companyID)->emit("update_online_count", $redis->hlen($key));
                }
            });
        });

        // 開始進(jìn)程時(shí)，監(jiān)聽2121端口，用戶數(shù)據(jù)推送
        $io->on("workerStart", function () use ($io) {
            /** @var Connection $redis */
            $redis = Yii::$app->redis;

            $httpWorker = new Worker("http://0.0.0.0:2121");
            // 當(dāng)http客戶端發(fā)來數(shù)據(jù)時(shí)觸發(fā)
            $httpWorker->onMessage = function ($conn, $data) use ($io, $redis) {
                $_POST = $_POST ? $_POST : $_GET;
                switch (@$_POST["action"]) {
                    case "message":
                        $to = "uid:" . @$_POST["receiverID"];
                        $_POST["content"] = htmlspecialchars(@$_POST["content"]);
                        $_POST["title"] = htmlspecialchars(@$_POST["title"]);

                        // 有指定uid則向uid所在socket組發(fā)送數(shù)據(jù)
                        if ($to) {
                            $io->to($to)->emit("new_msg", $_POST);
                        }

                        $companyID = @$_POST["companyID"];
                        $key = "socketio:company:" . $companyID;
                        $field = "uid:" . $to;

                        // http接口返回，如果用戶離線socket返回fail
                        if ($to && $redis->hget($key, $field)) {
                            return $conn->send("offline");
                        } else {
                            return $conn->send("ok");
                        }
                        break;
                    case "dynamic":
                        $to = "company:" . @$_POST["companyID"];
                        $_POST["content"] = htmlspecialchars(@$_POST["content"]);
                        $_POST["title"] = htmlspecialchars(@$_POST["title"]);

                        // 有指定uid則向uid所在socket組發(fā)送數(shù)據(jù)
                        if ($to) {
                            $io->to($to)->emit("update_dynamic", $_POST);
                        }

                        $companyID = @$_POST["companyID"];
                        $key = "socketio:company:" . $companyID;

                        // http接口返回，如果用戶離線socket返回fail
                        if ($to && $redis->hlen($key)) {
                            return $conn->send("offline");
                        } else {
                            return $conn->send("ok");
                        }
                }

                return $conn->send("fail");
            };

            // 執(zhí)行監(jiān)聽
            $httpWorker->listen();
        });

        // 運(yùn)行所有的實(shí)例
        global $argv;
        array_shift($argv);

        if (isset($argv[2]) && $argv[2] == "daemon") {
            $argv[2] = "-d";
        }

        Worker::runAll();
    }

    /**
     * 新建任務(wù)
     * @param $projectID
     * @param $categoryID
     * @return string
     * @throws ForbiddenHttpException
     * @since 2018-01-26
     */
    public function actionCreate($projectID, $categoryID)
    {
        $userID = Yii::$app->user->id;

        if (!Yii::$app->user->can("createTask")) {
            throw new ForbiddenHttpException(ResponseUtil::$msg[1]);
        }

        if ($data = Yii::$app->request->post()) {
            if (empty($data["name"])) {
                ResponseUtil::jsonCORS(["status" => Conf::FAILED, "msg" => "任務(wù)標(biāo)題不能為空"]);
            }

            // 保存任務(wù)
            $taskID = TaskService::factory()->save([
                "name"       => $data["name"],
                "creatorID"  => $userID,
                "companyID"  => $this->companyID,
                "level"      => $data["level"],
                "categoryID" => $categoryID,
                "projectID"  => $projectID,
                "content"    => $data["content"]
            ]);

            if ($taskID) {
                $url = Url::to(["task/view", "taskID" => $taskID]);
                $portrait = UserService::factory()->getUserPortrait($userID);
                $username = UserService::factory()->getUserName($userID);
                $title = "創(chuàng)建了新任務(wù)";
                $content = $data["name"];

                // 保存操作日志
                LogService::factory()->saveHandleLog([
                    "objectID"   => $taskID,
                    "companyID"  => $this->companyID,
                    "operatorID" => $userID,
                    "objectType" => Conf::OBJECT_TASK,
                    "content"    => $content,
                    "url"        => $url,
                    "title"      => $title,
                    "portrait"   => $portrait,
                    "operator"   => $username
                ]);

                // 動(dòng)態(tài)推送
                MsgService::factory()->push("dynamic", [
                    "companyID"  => $this->companyID,
                    "operatorID" => $userID,
                    "operator"   => $username,
                    "portrait"   => $portrait,
                    "title"      => $title,
                    "content"    => $content,
                    "url"        => $url,
                    "date"       => date("Y-m-d H:i:s"),
                ]);

                ResponseUtil::jsonCORS(null, Conf::SUCCESS, "創(chuàng)建成功");
            } else {
                ResponseUtil::jsonCORS(null, Conf::FAILED, "創(chuàng)建失敗");
            }
        } else {
            return $this->render("create", [
                "projectID" => $projectID,
                "category"  => TaskCategory::findOne(["id" => $categoryID]),
            ]);
        }
    }

LogService::factory()->saveHandleLog($args)

companyID 用于查詢該公司下的動(dòng)態(tài)

objectType和objectID組合可以查詢某個(gè)對象的操作日志，例如一個(gè)任務(wù)被操作的日志

receiverID和typeID和isRead組合可以查詢用戶未讀消息

MsgService::factory()->push($action, $args)