Thinkphp5.1 + Swoole 主动推送通知消息

1、安装swoole，之前已经讲过，就不重复了。

2、安装think-swoole扩展。

安装教程：https://www.kancloud.cn/manual/thinkphp5_1/675277；

3、新建Swoole.php文件：如图所示：

修改config目录下的swoole_server.php文件

'swoole_class' => 'app\index\controller\Swoole', // 自定义服务类名称

Swoole.php文件内容：

<?php

namespace app\index\controller;

use think\swoole\Server;

use think\Controller;

use think\Db;

require_once __DIR__ . '../../../../thinkphp/base.php';//加载base文件，同理于public/index.php的加载

//注意：只加载一次 require_once,也可以加载自定义的公共文件如：common.php

//require_once __DIR__ . '../../../../application/common/common.php';

class Swoole extends Server

{

protected $host = '0.0.0.0'; //监听所有地址

protected $port = 9502; //监听9502端口

protected $serverType = 'ssl';

   protected $mode = SWOOLE_PROCESS;

   protected $sockType = SWOOLE_SOCK_TCP | SWOOLE_SSL;

protected $option = [

'worker_num'=> 4, //设置启动的Worker进程数

'daemonize' => false, //守护进程化（上线改为true）

'backlog' => 128, //Listen队列长度

'dispatch_mode' => 2, //固定模式，保证同一个连接发来的数据只会被同一个worker处理

//心跳检测：每60秒遍历所有连接，强制关闭10分钟内没有向服务器发送任何数据的连接

'heartbeat_check_interval' => 60,

'heartbeat_idle_time' => 600,

 'ssl_cert_file' => '/www/server/nginx/conf/cert/chengzhigang.cn.crt',//证书地址

     'ssl_key_file' => '/www/server/nginx/conf/cert/chengzhigang.cn.key',//证书地址

];

public function onWorkerStart($server,$worker_id)

{

//$redis = sentinelPort();

// if($worker_id==0){

//     swoole_timer_tick(2000, function ($timer_id) use ($server,$redis) {

//         echo "执行定时任务：".date('Y-m-d H:i:s')."\n";

//         $fds = $redis->sMembers('noticeFds');

//         $time = time()-2;//两秒前的时间

//         $time = date('Y-m-d H:i:s',$time);

//         $where = [];

//         $where[] = ['create_time','>',$time];

//         $notice = $this->getNotice($where);

//         foreach($fds as $fd){

//             $server->push($fd, json_encode(['status'=>1,'msg'=>'推送成功','data'=>$notice]));

//         }

//     });

// }

}

public function onRequest($request,$response){

      $token = $request->post['token'];

      $info = $request->post['info'];

      $data = [];

      if($token==''){

          $data[] = $info;

          $redis = sentinelPort();

          $fds = $redis->sMembers('noticeFds');

foreach($fds as $fd){

//$this->swoole是Server.php中定义的

$this->swoole->push($fd, json_encode(['status'=>1,'msg'=>'推送成功','data'=>$data]));

}

}

}

//建立连接时回调函数

public function onOpen($server,$req)

{

echo "标识{$req->fd}建立了连接\n";

//获取访客来源

      $redis = sentinelPort();

$fd = $req->fd;

$redis->sAdd('noticeFds',$fd);

$notice = $this->getNotice(['enable'=>1]);

$server->push($fd, json_encode(['status'=>1,'msg'=>'连接成功','data'=>$notice]));

}

public function getNotice($where=[])

{

$notice = Db::name('admin_notice')->where($where)->field('msg,link')->order('sort')->select();

return $notice;

}

//接收数据时回调函数

public function onMessage($server,$frame)

{

}

//连接关闭时回调函数

public function onClose($server,$fd)

{

echo "标识{$fd}关闭了连接\n";

$redis = sentinelPort();

$redis->sRem('noticeFds', $fd);

}

}

4、文件配置说明：

1）、默认连接方式是socket,可以查看Server.php,文件地址：

\vendor\topthink\think-swoole\src\Server.php

因为我项目配有https，所以我自定义了一个连接方式为$serverType = 'ssl';

然后在Server.php中新增一个连接方式，也是websocket，如下：

// 实例化 Swoole 服务

switch ($this->serverType) {

case 'socket':

$this->swoole = new Websocket($this->host, $this->port, $this->mode, $this->sockType);

break;

case 'http':

$this->swoole = new HttpServer($this->host, $this->port, $this->mode, $this->sockType);

break;

case 'ssl':

               $this->swoole = new Websocket($this->host, $this->port, $this->mode, $this->sockType);

               break;

default:

$this->swoole = new SwooleServer($this->host, $this->port, $this->mode, $this->sockType);

}

然后跟http不同之处我都用红色标出来了。

2）、后台实现逻辑讲解：

在onOpen方法中将客户端fd保存到redis中，关于sentinelPort()这个方法可以看我之前的文档，这里就不讲了。

连接成功后将所有的通知消息返给客户端，并且在onClose方法中将客户端fd删除掉。

第一种思路：在onWorkStart方法中给第一个进程加入定时任务，定时查询数据库，并将最新的消息返给当前在线的所有客户端。

第二种思路：后台新建消息的时候触发且推送给当前在线的所有客户端。

当然，第二种思路是很完美的，我也正是用了第二种方法，具体实现原理：

1、后台新增消息成功后调用pushNotice方法，token是自定义的签名，$data是要推送的消息

public function pushNotice($token,$data){

$url = "https://www.chengzhigang.cn:9502";

httpsRequest($url,['token'=>$token,'info'=>$data]);

}

注意：$url 是用域名加端口号去连接，因为本项目是https的。

function httpsRequest($url, $data = null,$headers=[]) {

$curl = curl_init();

curl_setopt($curl, CURLOPT_URL, $url);

curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, FALSE);

curl_setopt($curl, CURLOPT_SSL_VERIFYHOST, FALSE);

if (!empty($data)) {

curl_setopt($curl, CURLOPT_POST, 1);

curl_setopt($curl, CURLOPT_POSTFIELDS, http_build_query($data));

}

curl_setopt($curl, CURLOPT_RETURNTRANSFER, 1);

curl_setopt($curl, CURLOPT_HTTPHEADER,["Accept: application/json"]);

if(!empty($headers)){

$headers[] = "Accept: application/json";

curl_setopt($curl, CURLOPT_HTTPHEADER,$headers);

}else{

curl_setopt($curl, CURLOPT_HTTPHEADER,["Accept: application/json"]);

}

$output = curl_exec($curl);

curl_close($curl);

if(is_null(json_decode($output))){

return $output;

}else{

return json_decode($output,true);

}

}

2、接受数据：用swoole中的onRequest方法，

5、客户端调用方法示例：

var wsServer = 'wss://www.chengzhigang.cn:9502';

var websocket = new WebSocket(wsServer);

websocket.onopen = function (evt) {

console.log(evt);

};

websocket.onclose = function (evt) {

console.log("Disconnected");

};

websocket.onmessage = function (evt) {

var data = JSON.parse(evt.data);

//处理数据

};

websocket.onerror = function (evt, e) {

console.log('Error occured: ' + evt.data);

};

6、命令执行：

在项目目录下面执行：php think swoole:server 可能会报错：

你的9502端口号已经被使用，如果你不想换端口号，需要先杀死9502的PID，然后重新启动，命令如下：

查看9502端口：lsof -i:9502

kill 16065   //杀死进程

然后打开网址连接socket,可能为报错：

原因有多种：1、端口号不一致；

2、端口号未开放；

3、网站是https的，需要用域名连接；

4、服务没有开启；

端口未开放解决方法：

查看开放的端口：

firewall-cmd --list-ports

  添加开放端口：

firewall-cmd --zone=public --add-port=9502/tcp --permanent

  重启防火墙：

systemctl restart firewalld.service

7、效果示例：

网址：https://www.chengzhigang.cn