摘要：此接口主要用于客服等有人工消息處理環(huán)節(jié)的功能，方便開(kāi)發(fā)者為用戶提供更加優(yōu)質(zhì)的服務(wù)。不支持廣告等營(yíng)銷(xiāo)類(lèi)消息以及其它所有可能對(duì)用戶造成騷擾的消息。

關(guān)于群發(fā)接口
1.訂閱號(hào)每天可以群發(fā)消息一條，服務(wù)號(hào)每月（自然月）四條的群發(fā)權(quán)限。開(kāi)發(fā)者模式下，可以通過(guò)高級(jí)群發(fā)接口，實(shí)現(xiàn)更靈活的群發(fā)能力。
2.注意
● 對(duì)于認(rèn)證訂閱號(hào)，群發(fā)接口每天可成功調(diào)用1次，此次群發(fā)可選擇發(fā)送給全部用戶或某個(gè)標(biāo)簽；
● 對(duì)于認(rèn)證服務(wù)號(hào)雖然開(kāi)發(fā)者使用高級(jí)群發(fā)接口的每日調(diào)用限制為100次，但是用戶每月只能接收4條，無(wú)論在公眾平臺(tái)網(wǎng)站上，還是使用接口群發(fā)，用戶每月只能接收4條群發(fā)消息，多于4條的群發(fā)將對(duì)該用戶發(fā)送失??；
● 具備微信支付權(quán)限的公眾號(hào)，在使用群發(fā)接口上傳、群發(fā)圖文消息類(lèi)型時(shí)，可使用a標(biāo)簽加入外鏈；
關(guān)于客服消息和模板消息接口
當(dāng)用戶和公眾號(hào)產(chǎn)生特定動(dòng)作的交互時(shí)（具體動(dòng)作列表請(qǐng)見(jiàn)下方說(shuō)明），微信將會(huì)把消息數(shù)據(jù)推送給開(kāi)發(fā)者，開(kāi)發(fā)者可以在一段時(shí)間內(nèi)（目前修改為48小時(shí)）調(diào)用客服接口，通過(guò)POST一個(gè)JSON數(shù)據(jù)包來(lái)發(fā)送消息給普通用戶。此接口主要用于客服等有人工消息處理環(huán)節(jié)的功能，方便開(kāi)發(fā)者為用戶提供更加優(yōu)質(zhì)的服務(wù)。
模板消息僅用于公眾號(hào)向用戶發(fā)送重要的服務(wù)通知，只能用于符合其要求的服務(wù)場(chǎng)景中，如信用卡刷卡通知，商品購(gòu)買(mǎi)成功通知等。不支持廣告等營(yíng)銷(xiāo)類(lèi)消息以及其它所有可能對(duì)用戶造成騷擾的消息。

簡(jiǎn)單的描述一下業(yè)務(wù)的背景，目前是做一個(gè)供求的微信公眾號(hào)，每當(dāng)用戶付費(fèi)發(fā)送一次供求的信息，我們需要將本條消息推送給所有的用戶，這樣可以讓用戶及時(shí)收到消息，實(shí)現(xiàn)消息的實(shí)時(shí)性和保證消息的有效性。但是群發(fā)的接口根本不能滿足我們的需求，于是我們利用模板消息和客服消息的接口來(lái)實(shí)現(xiàn)我們的需求，在此，有的人可能會(huì)發(fā)問(wèn)，為什么客服消息的接收要求這么{{BANNED}}還要利用它呢，其實(shí)是因?yàn)?，群發(fā)的消息在某些設(shè)備上收到的時(shí)候，就像微信好友發(fā)送的一條信息，更加吸引用戶去關(guān)注。
另外，我們還需要知道，假如我們的用戶用成千上萬(wàn)的人話，那么群發(fā)的方式是十分的耗時(shí)的，微信支付提供了notify的異步通知，之前我也一直嘗試?yán)眠@個(gè)現(xiàn)場(chǎng)的異步通知來(lái)實(shí)現(xiàn)群發(fā)，但是根據(jù)實(shí)際的使用，這個(gè)異步IO的閾值差不多在6分鐘左右，可是實(shí)際上發(fā)送一次模板的網(wǎng)絡(luò)耗時(shí)加上數(shù)據(jù)庫(kù)的IO其實(shí)還是時(shí)間還是挺長(zhǎng)的，因此我們不得不使用異步多線程來(lái)實(shí)現(xiàn)我們的群發(fā)的功能這里我利用的是swoole的擴(kuò)展來(lái)完成這一需求的。
swoole的異步郵件群發(fā)的demo，大家可以參照我的另一篇文章thinkphp5+swoole實(shí)現(xiàn)異步郵件群發(fā)（SMTP方式）這里可以較為詳細(xì)的了解服務(wù)端和客戶端的構(gòu)建。

centos7
swoole2.0+
tp5.0+
郵件發(fā)送
crontab定時(shí)任務(wù)

我這里就直接貼代碼了,需要注意的地方都寫(xiě)在了相關(guān)代碼的注釋?zhuān)⒁饪匆幌?/p>

/**
     * description:服務(wù)端
     */
    public function asyncSend()
    {
        $serv = new swoole_server("0.0.0.0", 9090);
        $serv->set(array(
            "task_worker_num" => 60,
            "daemonize" => true,
            "log_file" => "/var/www/html/myswl/tp/swoole.log"
            )
        );
        $serv->on("receive", function ($serv, $fd, $from_id, $data) {
            $task_id = $serv->task($data);
            echo "開(kāi)始投遞異步任務(wù) id=$task_id
";
        });

        $serv->on("task", function ($serv, $task_id, $from_id, $data) {
            echo "接收異步任務(wù)[id=$task_id]" . PHP_EOL;
            $data = json_decode($data,true);

//這里要特別的說(shuō)明，我這里用到了數(shù)據(jù)庫(kù)的遠(yuǎn)程鏈接，因?yàn)橹Ц兜墓δ茉诹硪慌_(tái)window虛擬云主機(jī)上的，所以不得不利用遠(yuǎn)程訪問(wèn)的方式。Db::connect($conn,true);的第二個(gè)參數(shù)給給予關(guān)注，因?yàn)槲覀儧](méi)發(fā)送一次其實(shí)都會(huì)去進(jìn)行一次遠(yuǎn)程數(shù)據(jù)庫(kù)連接，所以頻繁的連接中，肯定會(huì)有連接失敗的情況，因此我們需要做好斷線重連的配置
            $conn = "mysql://用戶名:數(shù)據(jù)庫(kù)遠(yuǎn)程鏈接地址:3306/密碼#utf8";
            $db = Db::connect($conn,true);

            $now = date("Y-m-d H:i:s");
            $users = $db
                ->table("tp_receiver")
                ->field("openid")
                ->where("(`expire_time` > "{$now}" OR `send_status` = 1 ) AND `receive_status` = 1 ")
                ->limit($data["flag"]*500,500)
                ->select();

            echo $db->getLastSql();

            echo "send start--" . date("H:i:s") . PHP_EOL;
            foreach ($users as $user) {
                $token = $db->table("tp_accesstoken")->field("accesstoken")->find();
//這里是客服消息
                $templData2 = array(
                    "touser" => $user["openid"],
                    "msgtype" => "text",
                     "text" => array("content" => $data["content"]."

點(diǎn)擊下方“信息查詢”查看更多求購(gòu)信息。"."
回復(fù)0取消接收信息，回復(fù)1重新接收信息")
                );

                $res = $this->sendCustomMessage($templData2,$token["accesstoken"]);
                $res = json_decode($res, true);
//判斷客服消息是否成功發(fā)送
                if($res["errcode"] == 45047 || $res["errcode"] == 45015) {
                    $templData = array(
                        "touser" => $user["openid"],
                        "template_id" => "模板消息",
                        "url" => "",
                        "data" => array(
                            "first" => array("value" => "您好，您收到一條新的提醒", "color" => "#173177"),
                            "keyword1" => array("value" => "求購(gòu)信息", "color" => "#173177"),
                            "keyword2" => array("value" => $data["content"], "color" => "#173177"),
                            "keyword3" => array("value" => $data["phone"], "color" => "#173177"),
                            "keyword4" => array("value" => date("Y-m-d H:i:s"), "color" => "#173177"),
                            "remark" => array("value" => "點(diǎn)擊下方“信息查詢”查看更多求購(gòu)信息。"."回復(fù)0取消接收信息，回復(fù)1重新接收信息", "color" => "#173177")
                        ),
                    );
                    $res = $this->sendTemplateMessage($templData, $token["accesstoken"]);
                    Log::write("send to".$user["openid"].$res . PHP_EOL);
                }
            }
            echo "send end--" . date("H:i:s") . PHP_EOL;

            $serv->finish("");
        });

        $serv->on("finish", function ($serv, $task_id, $data) {
            echo "finish time--" . microtime(true) . PHP_EOL;
            echo "異步任務(wù)[id=$task_id]完成" . PHP_EOL;
        });

        $serv->start();
    }

然后我們?cè)贑LI模式下進(jìn)入項(xiàng)目的根目錄，執(zhí)行

php public/index.php demo/wechat/asyncSend

這樣我們的服務(wù)端就以守護(hù)進(jìn)程的模式一直運(yùn)行來(lái)我們的后臺(tái)了，通過(guò)ps -aux | grep asyncSend
可以看見(jiàn)，已經(jīng)有62個(gè)進(jìn)程在處于S（睡眠待喚醒）的狀態(tài)了,除了60個(gè)task進(jìn)程還用一個(gè)master和一個(gè)woker進(jìn)程。

代碼如下，需要注意的地方都寫(xiě)在了相關(guān)代碼的注釋?zhuān)⒁饪匆幌?/p>

/**
     * description:客戶端
     */
    public function index()
    {
//因?yàn)檫@個(gè)群發(fā)比較敏感，我們需要做一個(gè)token的機(jī)制，我這邊就用最簡(jiǎn)單的發(fā)送方和接收方都以明文的方式來(lái)做了。
            $token = $_GET["token"];
            if($token != "test"){
                exit;
            }
//content是發(fā)送的內(nèi)容，因?yàn)椴豢深A(yù)估里面的東西，所以進(jìn)行加解密
            $content = rawurldecode($_GET["content"]);
            $flag = $_GET["flag"];
            $id = $_GET["id"];
            $phone = $_GET["phone"];

            $data["content"] = $content;
            $data["flag"] = $flag;
            $data["phone"] = $phone;

            Log::write(self::json_encode($data));

            $insert = [
                "flag"=>$flag,
                "miaomu_id"=>$id,
                "status"=>1,
                "createtime"=>date("Y-m-d H:i:s"),
                "content"=>$content
            ];
            Db::table("task")->insert($insert);

//異步客戶端
            $client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_SYNC);
            $ret = $client->connect("127.0.0.1", 9090);
//當(dāng)無(wú)法連接的時(shí)候，發(fā)送告警郵件
            if (empty($ret)) {
                SendMail::postmail("[email protected]","警告","error!connect to swoole_server failed");
                SendMail::postmail("[email protected]","警告","error!connect to swoole_server failed");
                Log::write("error!connect to swoole_server failed");
            } else {
                $client->send(self::json_encode($data));
            }
    }

這個(gè)群發(fā)已經(jīng)涉及到金額了，所以我們更要關(guān)系服務(wù)的運(yùn)行穩(wěn)定了，這里我們簡(jiǎn)單的利用crontab定時(shí)任務(wù)和Php的一些shell相關(guān)函數(shù)來(lái)實(shí)現(xiàn)端口的監(jiān)控。
本次用到的定時(shí)任務(wù)

*/1 * * * * curl http://你的域名/index.php/demo/Jrmm/checkPortStatus?token=test

就是實(shí)現(xiàn)一個(gè)每分鐘去執(zhí)行我們下面php代碼的一個(gè)任務(wù)，這里我沒(méi)有直接用shell來(lái)操作，原因有3點(diǎn)，1是我不是很熟悉shell命令，2是我們不太熟悉shell命令，3是寫(xiě)在php里面更方便我去寫(xiě)相關(guān)代碼和利用已有的一些方法，比如郵件發(fā)送。這樣雖然多了一點(diǎn)網(wǎng)絡(luò)資源的消耗，但是也還劃算。
具體的監(jiān)控代碼，這邊實(shí)現(xiàn)的時(shí)候會(huì)出現(xiàn)很多權(quán)限的問(wèn)題，我就不多說(shuō)了，遇到的時(shí)候自行百度。
/**

 * description:8082服務(wù)端口監(jiān)控
 */
public function checkPortStatus(){
    if (!isset($_GET["token"]) || $_GET["token"] != "test"){
        exit();
    }
    $res1 = exec("sudo netstat -lpn | grep 9090");
    Log::write($res1);
    if($res1 == ""){
        Log::write("9090stop");
        SendMail::postmail("[email protected]","警告","9090端口服務(wù)錯(cuò)誤");
        SendMail::postmail("[email protected]","警告","9090端口服務(wù)錯(cuò)誤");

//重啟我們的服務(wù)端,這里需要注意的是，我沒(méi)有用到swoole提供的平滑重啟的功能，很可能會(huì)造成數(shù)據(jù)的丟失，這別額外的需要注意

        exec("sudo php /var/www/html/myswl/tp/public/index.php demo/jrmm/asyncSend");
        $res = exec("sudo netstat -lpn | grep 8082");
        if($res != ""){
            Log::write("9090restart success");
            SendMail::postmail("[email protected]","警告解除","9090端口重啟成功");
            SendMail::postmail("[email protected]","警告解除","9090端口重啟成功");
        }
    }
}

我們利用PHP curl函數(shù)來(lái)模擬一次支付成功后調(diào)用我們?nèi)喊l(fā)的功能。

$content = "test";
        for ($j=0;$j<3;$j++){
             $url = "http://你的域名/index.php/demo/Jrmm/index"."?flag=".$j."&id=1."&token=test"."&phone="110&content=".rawurlencode($content);
            $this->http_post($url);
        }
function http_post($url){
    $ch = curl_init();
    curl_setopt($ch, CURLOPT_URL, $url);
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
    curl_setopt($ch, CURLOPT_HEADER, 0);
    $res= curl_exec($ch);
    curl_close($ch);
    return $res;
}

為什么循環(huán)三次呢，因?yàn)槲覀兺ㄟ^(guò)測(cè)試發(fā)送發(fā)送1500次消息的時(shí)候，耗時(shí)差不多6分鐘，但是我們的項(xiàng)目的并發(fā)很低，那么就無(wú)法充分利用我們開(kāi)啟的60個(gè)task進(jìn)程，所以我們將1500分成三次去發(fā)送那么實(shí)際上我們消耗了幾乎可以忽略不計(jì)的網(wǎng)絡(luò)消耗，讓我們的發(fā)送的性能提高了三倍多，實(shí)際的項(xiàng)目中，發(fā)送1500多條實(shí)際耗時(shí)只要不到兩分鐘。當(dāng)然當(dāng)并發(fā)量更大的時(shí)候，我們還可以采用隊(duì)列的方式來(lái)處理，這樣需要我們隊(duì)task進(jìn)程管理更加的熟練。

收到的推送消息

告警

發(fā)送的日志

http://pan.baidu.com/s/1c2q7ik4