Swoole实现Websocket每秒钟主动推送

Swoole实现Websocket主动推送

需求&＃xff1a;
需要每秒钟向前台推送一次行情数据。
向某个用户推送消息
及时响应前端发出的请求
握手时效验令牌
每个ip 不超过十个连接
因为网上没有很具体的相关文档&＃xff0c;只能摸着石头过河&＃xff0c;现功能已经大致实现&＃xff0c;如果有应用不合理的地方希望大神不吝赐教&＃xff01;

$config &＃61; require(&＃39;/config/swoole.php&＃39;);
class Websocket {
public $server;
private $config;
public function __construct($config) {
$this->config &＃61; $config;
$server &＃61; new Swoole\WebSocket\Server($this->config[&＃39;web_host&＃39;],$this->config[&＃39;web_port&＃39;]);
$this->server &＃61; $server;
$this->server->set([
&＃39;worker_num&＃39; &＃61;> 8,
&＃39;daemonize&＃39; &＃61;> 1,
&＃39;backlog&＃39; &＃61;> 128,
&＃39;max_request&＃39; &＃61;> 10000,
&＃39;heartbeat_check_interval&＃39; &＃61;> 60,
&＃39;task_worker_num&＃39; &＃61;> 24,
]);
$process &＃61; new Swoole\Process(function($process) use ($server) {
while (true) {
$show_market_list &＃61; $this->getRealMarket();
if ($show_market_list) {
foreach ($this->server->connections as $conn) {
$this->server->push($conn,json_encode([&＃39;type&＃39; &＃61;> &＃39;market&＃39;,&＃39;date&＃39; &＃61;> $show_market_list]));
}
}
sleep(1);
}
});
$newprocess &＃61; new Swoole\Process(function($process) use ($server){
while (true) {
$redis &＃61; $this->newRedis();
$new_list &＃61; $redis->hGetALL(&＃39;swoole_new_push&＃39;);
if ($new_list) {
//把消息推给所有用户
foreach ($new_list as $k &＃61;> $v) {
//查找用户的fd 检查用户是否在线
$user_fd &＃61; $redis -> hGet(&＃39;swoole_uid_fd&＃39;,$k);
if ($user_fd) {
$this->server->push($user_fd,json_encode([&＃39;type&＃39; &＃61;> &＃39;new&＃39;,&＃39;date&＃39; &＃61;> $v]));
}
}
}
sleep(1);
}
});
$this->server->addProcess($newprocess);
$this->server->addProcess($process);
$this->server->on(&＃39;handshake&＃39;,function($request,$response){
$secWebSocketKey &＃61; $request->header[&＃39;sec-websocket-key&＃39;];
$patten &＃61; &＃39;#^[&＃43;/0-9A-Za-z]{21}[AQgw]&＃61;&＃61;$#&＃39;;
if (0 &＃61;&＃61;&＃61; preg_match($patten, $secWebSocketKey) || 16 !&＃61;&＃61; strlen(base64_decode($secWebSocketKey))) {
$response->end();
$this->server->close($request->fd);
return 500;
}
$key &＃61; base64_encode(sha1(
$request->header[&＃39;sec-websocket-key&＃39;] . &＃39;258EAFA5-E914-47DA-95CA-C5AB0DC85B11&＃39;,
true
));
$headers &＃61; [
&＃39;Upgrade&＃39; &＃61;> &＃39;websocket&＃39;,
&＃39;Connection&＃39; &＃61;> &＃39;Upgrade&＃39;,
&＃39;Sec-WebSocket-Accept&＃39; &＃61;> $key,
&＃39;Sec-WebSocket-Version&＃39; &＃61;> &＃39;13&＃39;,
];
if (isset($request->header[&＃39;sec-websocket-protocol&＃39;])) {
//占用ip
$real_ip &＃61; $request->header[&＃39;x-real-ip&＃39;];
$result &＃61; $this->checkMac(md5($real_ip));
if (!$result) {
$response->end();
$this->server->close($request->fd);
return 500;
}
$headers[&＃39;Sec-WebSocket-Protocol&＃39;] &＃61; $request->header[&＃39;sec-websocket-protocol&＃39;];
}else{
$response->end();
$this->server->close($request->fd);
return 500;
}
foreach ($headers as $key &＃61;> $val) {
$response->header($key, $val);
}
$response->status(101);
$response->end();
$getAllMarket &＃61; $this->getAllMarket();
$this->server -> push($request->fd,json_encode($getAllMarket));
});
$this->server->on(&＃39;open&＃39;, function (swoole_websocket_server $server, $request) {

});$this->server->on(&＃39;task&＃39;,function(swoole_server $server, $task_id, $from_id, $data){$task &＃61; json_decode($data[&＃39;type&＃39;],1);switch ($task[&＃39;type&＃39;]) {//K线case &＃39;kline&＃39;:$result[&＃39;type&＃39;] &＃61; $task[&＃39;type&＃39;];$result[&＃39;kline_time_type&＃39;] &＃61; $task[&＃39;kline_time_type&＃39;];$result[&＃39;market&＃39;] &＃61; $task[&＃39;market&＃39;];$result[&＃39;data&＃39;] &＃61; $this->getKlineDate($task[&＃39;kline_time_type&＃39;],$task[&＃39;market&＃39;]);$result[&＃39;history&＃39;] &＃61; $this->getMarketRecord($task[&＃39;market&＃39;]);$server -> push($data[&＃39;fd&＃39;],json_encode($result));break;//用户注册case &＃39;login&＃39;:$token &＃61; $task[&＃39;token&＃39;];$redis &＃61; $this->newRedis();$uid &＃61; $redis->hGet(&＃39;user_token&＃39;,$token);if ($uid) {//找到了 用户的uid$redis -> hSet(&＃39;swoole_uid_fd&＃39;,$uid,$data[&＃39;fd&＃39;]);$redis -> hSet(&＃39;swoole_fd_uid&＃39;,$data[&＃39;fd&＃39;],$uid);$result[&＃39;type&＃39;] &＃61; &＃39;login&＃39;;$result[&＃39;code&＃39;] &＃61; 200;}else{$result[&＃39;type&＃39;] &＃61; &＃39;login&＃39;;$result[&＃39;code&＃39;] &＃61; 400;}$server -> push($data[&＃39;fd&＃39;],json_encode($result));break;//响应消息case &＃39;respond&＃39;:$token &＃61; $task[&＃39;token&＃39;];$redis &＃61; $this->newRedis();$uid &＃61; $redis->hGet(&＃39;user_token&＃39;,$token);if ($uid) {$redis->hDel(&＃39;swoole_new_push&＃39;,$uid);$result[&＃39;type&＃39;] &＃61; &＃39;respond&＃39;;$result[&＃39;code&＃39;] &＃61; 200;}else{$result[&＃39;type&＃39;] &＃61; &＃39;respond&＃39;;$result[&＃39;code&＃39;] &＃61; 400;}$server -> push($data[&＃39;fd&＃39;],json_encode($result));break;default:break;}});$this->server->on(&＃39;message&＃39;, function (Swoole\WebSocket\Server $server, $frame) {$task[&＃39;type&＃39;] &＃61; $frame->data;$task[&＃39;fd&＃39;] &＃61; $frame->fd;$server -> task($task);});$this->server->on(&＃39;close&＃39;, function ($server, $fd) {//删除用户在线$redis &＃61; $this->newRedis();$uid &＃61; $redis -> hGet(&＃39;swoole_fd_uid&＃39;,$fd);if ($uid) {$redis -> hDel(&＃39;swoole_uid_fd&＃39;,$uid);$redis -> hDel(&＃39;swoole_fd_uid&＃39;,$fd);}});$this->server->on(&＃39;request&＃39;, function ($request, $response) {});$this->server->start();
}
//K线数据
public function getKlineDate($type,$market){$redis &＃61; $this->newRedis();$date &＃61; $redis->hGet($type,$market);return $date;
}
public function getMarketRecord($market){$redis &＃61; $this->newRedis();$result[&＃39;yest_close_price&＃39;] &＃61; $redis->hget(&＃39;yest_close_price&＃39;,$market);$result[&＃39;today_open_price&＃39;] &＃61; $redis->hget(&＃39;today_open_price&＃39;,$market);$result[&＃39;today_low_price&＃39;] &＃61; $redis->hget(&＃39;today_low_price&＃39;,$market);$result[&＃39;today_high_price&＃39;] &＃61; $redis->hget(&＃39;today_high_price&＃39;,$market);return $result;
}
//获取全部行情
public function getAllMarket(){$redis &＃61; $this->newRedis();$market_buy_list &＃61; $redis -> HgetAll(&＃39;real_market_buy&＃39;);$market_sell_list &＃61; $redis -> HgetAll(&＃39;real_market_sell&＃39;);$market_real_list &＃61; $redis->HgetAll(&＃39;real_market&＃39;);foreach ($market_real_list as $k &＃61;> $v) {$show_market_list[$k][&＃39;real&＃39;] &＃61; $v;$show_market_list[$k][&＃39;sell&＃39;] &＃61; $market_sell_list[$k];$show_market_list[$k][&＃39;buy&＃39;] &＃61; $market_buy_list[$k];}return $show_market_list;
}//获取变动的行情 进行广播
public function getRealMarket(){$redis &＃61; $this->newRedis();$market_buy_list &＃61; $redis -> HgetAll(&＃39;real_market_buy&＃39;);$market_sell_list &＃61; $redis -> HgetAll(&＃39;real_market_sell&＃39;);$market_real_list &＃61; $redis -> HgetAll(&＃39;real_market&＃39;);foreach ($market_real_list as $k &＃61;> $v) {$result[&＃39;real&＃39;] &＃61; $v;$result[&＃39;sell&＃39;] &＃61; $market_sell_list[$k];$result[&＃39;buy&＃39;] &＃61; $market_buy_list[$k];$history_market &＃61; $redis -> hGet(&＃39;history_market&＃39;,$k);if ($history_market &＃61;&＃61; json_encode($result)) {//这一次读取的数据 和历史数据是一样的continue;}else{$show_market_list[$k] &＃61; $result;}}return $show_market_list;
}
//每个IP 每分钟最多10个连接
public function checkMac($mac,$status &＃61; 0){$zero_time &＃61; strtotime(date(&＃39;Y-m-d H:i&＃39;));$redis &＃61; $this->newRedis();$res &＃61; $redis->hGet(&＃39;Conn_Mac&＃39;,$mac);if ($res) {try {$res &＃61; json_decode($res,1);if ($res[&＃39;time&＃39;] !&＃61; $zero_time) {//如果已经超过了一分钟区间$json[&＃39;time&＃39;] &＃61; $zero_time;$json[&＃39;rate&＃39;] &＃61; 1;$redis -> Hset(&＃39;Conn_Mac&＃39;,$mac,json_encode($json));return true;}else{if ($res[&＃39;rate&＃39;] <100) {//如果连接数还没有达到上限$json &＃61; json_encode([&＃39;time&＃39; &＃61;> $zero_time,&＃39;rate&＃39; &＃61;> $res[&＃39;rate&＃39;] &＃43; 1]);$redis ->Hset(&＃39;Conn_Mac&＃39;,$mac,$json);return true;}else{//连接次数达到上线return false;}}} catch (Exception $e) {$json[&＃39;time&＃39;] &＃61; $zero_time;$json[&＃39;rate&＃39;] &＃61; 1;$redis ->Hset(&＃39;Conn_Mac&＃39;,$mac,json_encode($json));return true;}}else{$arr[&＃39;time&＃39;] &＃61; $zero_time;$arr[&＃39;rate&＃39;] &＃61; 1;$redis -> hSet(&＃39;Conn_Mac&＃39;,$mac,json_encode($arr));return true;}
}
public function newRedis(){$redis &＃61; new redis();$redis -> connect($this->config[&＃39;read_redis_host&＃39;],$this->config[&＃39;read_redis_prot&＃39;]);$redis -> auth($this->config[&＃39;read_redis_auth&＃39;]);return $redis;
}

}
new Websocket($config);
?>