一、查看PHP的swoole扩展
打开phpinfo,搜索，没有安装的需要安装下
二、开始编写swoole服务代码：
问件目录：/console/controllers/WebSocketController.php
| 遇到问题1: 无法实现根据商户的uid给特定的商户推送消息，只能通过连接的用户进行所有用户推送 |
| 解决方案：1、客户端连接websocket服务的时候，传商户的uid,服务端client接收商户的uid，通过redis将商户的uid作为key，拼接固定前缀，客户端连接的fd为值，进行存储； 2、主动推送server连接服务端client主动推送消息的时候，json串中传要推送的商户的uid； 3、服务端message收到消息后，解析json传中传递的商户uid，通过uid的key值获取redis中存储的商户连接的fd值，通过fd值给商户推送消息； 4、解决根据商户uid给特定商户推送特定消息的问题； |
namespace console\controllers;
use \Swoole\WebSocket\Server;
use Yii;
use yii\console\Controller;
use yii\helpers\Json;
class WebSocketController extends Controller
{
/**
* @var Server
*/
public $serv;
public function actionRun()
{
//配置ssl加密连接 , SWOOLE_PROCESS, SWOOLE_SOCK_TCP|SWOOLE_SSL
$this->serv = new Server("0.0.0.0", 9503);
$this->serv->set([
'worker_num' => 10, //开启的进程数 ?般为cup核数 1-4 倍
'task_worker_num' => 10, //配置 Task 进程的数量。
'daemonize' => 1, //设置 daemonize => true 时，程序将转入后台作为守护进程运行。长时间运行的服务器端程序必须启用此项。如果不启用守护进程，当 ssh 终端退出后，程序将被终止运行。
'max_request' => 10000, //设置 worker 进程的最大任务数。
'dispatch_mode' => 5, //uid dispatch UID 分配 需要用户代码中调用 Server->bind() 将一个连接绑定 1 个 uid。然后底层根据 UID 的值分配到不同的 Worker 进程。
'open_eof_check' => true,
'package_eof' => PHP_EOL,
'log_file' => Yii::getAlias('@runtime/swoole_below_cps.log'),
'reload_async' => true, //开启的主要目的是为了保证服务重载时，协程或异步任务能正常结束。
//'enable_reuse_port' => true, //启用端口重用后，可以重复启动监听同一个端口的 Server 程序
'heartbeat_idle_time' => 600,
'heartbeat_check_interval' => 60,
// 'ssl_cert_file' => EnvHelper::get('SSL_PEM'), //ssl加密配置文件
// 'ssl_key_file' => EnvHelper::get('SSL_KEY'), //ssl加密配置key文件
]);
$this->serv->on('Start', [$this, 'onStart']);
$this->serv->on('Open', [$this, 'onOpen']);
$this->serv->on('Message', [$this, 'onMessage']);
$this->serv->on('Request', [$this, 'onRequest']);
$this->serv->on('Receive', [$this, 'onReceive']);
$this->serv->on('Task', [$this, 'onTask']);
$this->serv->on('Finish', [$this, 'onFinish']);
//dd('先不启');
$this->serv->start();
}
public function onStart(Server $server)
{
static::log('平台Websocket启动'.$server->worker_id . PHP_EOL);
}
/**
* Param: 当 WebSocket 客户端与服务器建立连接并完成握手后会回调此函数。
* User: 赫赫
* Date: 2022/6/13
* @param Server $server
* @param $request
* @return void
*/
public function onOpen(Server $server, $request) {
//接收uid
$uid = $request->get['uid'];
//$type = $request->get['type'];
static::log("活平台-建立连接: 用户uid= {$uid}, fd={$request->fd}建立连接\n");
if(!empty($uid)){
//将接收到的用户ID存储在redis中
$redisKey = "BELOWCPSWEBSOCKET:".$uid;
$redis = \Yii::$app->redis;
$res = $redis->set($redisKey,$request->fd);
static::log("平台-将接收到的用户ID存储在redis中:$res\n");
}
}
/** 当服务器收到来自客户端的数据帧时会回调此函数。
* Param:
* User: 赫赫
* Date: 2022/6/13
* @param Server $server
* @param $frame
* @return void
*/
public function onMessage(Server $server, $frame)
{
static::log('websocket服务收到老铁发的消息:' . $frame->data);
$data = json_decode($frame->data, true);
static::log('websocket服务收到老铁发的消息解密:' . $data['uid']."\n");
//根据uid获取redis中的fd,根据fd给用户推送消息
if(!empty($data['uid'])){
//获取redis中的用户fd
$redisKey = "BELOWCPSWEBSOCKET:".$data['uid'];
$redis = \Yii::$app->redis;
$u_fd = $redis->get($redisKey);
static::log("websocket服务获取老铁连接fd={$u_fd}发消息\n");
if($server->isEstablished($u_fd)){
$server->push($u_fd, $data['msg']);
}
}
// else{
// //没有uid
// foreach ($server->connections as $fd) {
// if ($server->isEstablished($fd)) {
// $server->push($fd, $data['msg']);
// }
// }
// }
}
/**
* Param: 接收http请求从get获取message参数的值，给用户推送
* User: 赫赫
* Date: 2022/6/13
* @param Server $serv
* @param $response
* @return void
*/
public function onRequest(Server $serv, $response){
$uid = $serv->get['uid'];
static::log('websocket服务收到:'.$uid);
foreach ($this->serv->connections as $fd) {
// 需要先判断是否是正确的websocket连接，否则有可能会push失败
if ($this->serv->isEstablished($fd)) {
$this->serv->push($fd, $serv->get['message']);
}
}
}
public function onReceive(Server $serv, $fd, $reactor_id, $data){
$conn = $serv->connection_info($fd);
echo "worker_id: " . $serv->worker_id . PHP_EOL;
if (empty($conn['uid'])) {
$uid = $fd + 1;
if ($serv->bind($fd, $uid)) {
$serv->send($fd, "bind {$uid} success");
}
} else {
if (!isset($serv->fdlist[$fd])) {
$serv->fdlist[$fd] = $conn['uid'];
}
print_r($serv->fdlist);
foreach ($serv->fdlist as $_fd => $uid) {
$serv->send($_fd, "{$fd} say:" . $data);
}
}
}
/**
* 处理任务
* @param \Swoole\Server $serv
* @param int $task_id
* @param int $from_id
* @param string $data
*/
public function onTask(Server $serv, int $task_id, int $from_id, string $data)
{
static::log('收到新的任务');
$data = Json::decode($data, true);
$serv->finish($data);
}
/**
* @param Server $server
* @param int $task_id
* @param mixed $data
*/
public function onFinish(Server $server, int $task_id, $data)
{
static::log('任务完成');
}
/**
* @param mixed $message
*/
private static function log($message)
{
$str = "[" . date('Y-m-d H:i:s') . '] ';
if (is_string($message)) {
$str .= $message;
}
if (is_array($message)) {
$str .= var_export($message, true);
}
$str .= PHP_EOL;
// echo $str;
\Yii::$app->byfLog->channel('BelowCpsWebSocket')->info($str);
}
}三、启动/停止
1、启动命令：php yii web-socket/run
2、停掉服务：根据端口号查看运行服务 ：netstat -nap | grep 9503
???????????????? kill id