swoole +tp5框架(thinkphp5+) 接入方法
本文系麦子时光原创，转载请注明出处，谢谢
https://www.jianshu.com/p/ff6ec1e143cb
不要用下面方法
看下 thinkphp 的入口文件 index.php
自动加载过程 非常的 溜 。。。。
我们自己建立一个 专门为 swoole 启动用的
入口文件 为了体现 这个文件放置的随意性 我们把
它建在 和 app 平级的目录中 ，且叫叫 tasktest.php
#!/usr/bin/env php
$default_module = 'core';
$default_controller = '';
$default_action = '';
$default_forbid_module = ['common','index','admin'];
//默认模块
define('__default_module__', $default_module);
//默认控制器
define('__default_controller__', $default_controller);
//默认方法
define('__default_action__', $default_action);
//默认禁止访问模块
define('__default_forbid_module__',json_encode($default_forbid_module));
// 上面这几步 可根据是否用到 默认 模块 等 进行配置 ，如无用到 可以省略
define('APP_PATH', __DIR__ . '/app/');
// 下面 这个最 主要的配置 就是 绑定 模块到 指定 swoole 启动的 文件 名字随意 对应就好
define('BIND_MODULE','core/Sio');
define('ROOT_PATH', __DIR__ . '/');
//前台模版目录
define('TPL_PATH', ROOT_PATH . 'tpl/');
//define("CONF_PATH",__DIR__.'/../conf/');
define("CONF_PATH",__DIR__.'/conf/');
define('NOW_TIME', $_SERVER['REQUEST_TIME']);
//// 加载框架引导文件
require __DIR__ . '/thinkphp/start.php';
这里我们 看下 对应的绑定模块 core /sio
// 下面是我自己的业务 代码
整体结构
构造函数
其他三个函数
触发相关 业务 这已经是swoole 内部的原理了 ，此处不赘述，只是这里可以利用框架的各种函数和简单的异步内存共享
由于动用了task
不要忘了finish
当然 还要用到clonse
onclonse 部分
现在 只要 运行
运行示例
欢迎 微信交流
本文系麦子时光原创，转载请注明出处，谢谢
https://www.jianshu.com/p/ff6ec1e143cb
sio 部分 源码分享
<?php
namespace app\core\controller;
use Swoole\Server;
use think\Controller;
use think\Db;
class Sioextends Controller
{
protected $port = 9052;
private $serv;
private $db_config = [];
private $redis_server = "127.0.0.1";
private $redis_port = "6379";
private $redis_pwd = "";
private $all_fd_token_map = "all_tunnel_online_map";
public function __construct()
{
/* 读取站点配置 */
$this->set_config();
echo "构造函数初始化。。。\n";
// var_dump(config("database"));
$this->db_config= config("database");
//redie 配置
$this->redis_server= !empty(config("PUBLIC_REDIS_ADDR"))? config("PUBLIC_REDIS_ADDR"): "127.0.0.1";
$this->redis_port= !empty(config("PUBLIC_REDIS_PORT"))? config("PUBLIC_REDIS_PORT"): "6379";
$this->redis_pwd= !empty(config("PUBLIC_REDIS_PWD"))? config("PUBLIC_REDIS_PWD"): "";
$this->clean_all_tunnel_key();
//swoole
$this->serv= new \swoole_server("0.0.0.0", $this->port);
$this->serv->set(array(
'worker_num' => 8,//建议开启的worker进程数为cpu核数的1-4倍
'daemonize' => false,
'max_request' => 10000,
'dispatch_mode' => 2,
'debug_mode' => 1,
'task_worker_num' => 8
));
//'reactor_num' => 8 //，默认会启用CPU核数相同的数量， 一般设置为CPU核数的1-4倍，最大不得超过CPU核数*4。
$this->serv->on('Start', array($this, 'onStart'));
$this->serv->on('Connect', array($this, 'onConnect'));
$this->serv->on('Receive', array($this, 'onReceive'));
$this->serv->on('Close', array($this, 'onClose'));
$this->serv->on('Task', array($this, 'onTask'));
// bind callback
$this->serv->on('Finish', array($this, 'onFinish'));
$this->serv->start();
if (!defined('GLOBAL_START')) {
$server = new Server();
define('GLOBAL_START', true);
}
}
public function onStart($serv)
{
echo "Start OK\n";
echo "确保 onstart 时 所有的 相关都初始化 ！重启后 fd 会重头再记录 ，redis 里面的数据 将失准";
dump(config("PUBLIC_REDIS_ADDR"));
// 清空已有 的redis 相关业务 可能涵盖 多平台
}
public function onConnect($serv, $fd, $from_id)
{
// $serv->send($fd, "Hello {$fd}!"); // 打招呼
echo "lingking——fd:----" . $fd; // 打印
echo " ";
echo "lingking——from_id:----" . $from_id; // 打印work id
}
public function onReceive(\swoole_server$serv, $fd, $from_id, $data)
{
echo "有新消息 来自客户端 {$fd} Client :{$data}\n";
if ($this->is_json($data)) {
$data = json_decode($data, true);
}
$param = array(
'fd' => $fd,
'data' => $data
);
// start a task
$serv->task(json_encode($param));
// echo "上面已经 交个task 这里不影响 做其他事 over\n";
}
public function onTask($serv, $task_id, $from_id, $in_data)
{
$backbool = "false";
// echo "This Task {$task_id} from Worker {$from_id}\n";
// echo "Data: {$in_data}\n";
// var_dump(json_decode($in_data,true));
$fd = json_decode($in_data, true)['fd'];
$data = json_decode($in_data, true)['data'];
if (!isset($data["token"])|| !isset($data["platform"])) {
echo "缺少token或者platform";
$serv->send($fd, "缺少token或者platform"); // 这里作为回复客户端
return "fd: {$fd} Task {$task_id}'s result";
}
// data 中 有三参数 token platfom info(内涵 now_mac set mac)
dump($this->redis_server);
$redis = new \Redis();
$redis->pconnect($this->redis_server, $this->redis_port);
if (!empty($this->redis_pwd)) {
$redis->auth($this->redis_pwd);
}
$tokenall = $data["token"];
$time_length = 3 * 60;
$bad_token_key = "aur_bad_token_" . $tokenall;
$check_bool = $this->bad_token_check($bad_token_key);
if (empty($check_bool)) {
$backbool = "false";
$re_mag = $this->send_msg($fd, $backbool); // 这里作为回复客户端
return "fd: {$fd} 触发的 Task {$task_id} 的 结果:{$re_mag}\n";
}
$platform = $data["platform"];
$token = substr($tokenall, 0, 32);
$tunnel_id = substr($tokenall, 33);
// 一个键 两个囊 一个放最大数量 另一个放 fd 对 用hash token_all:[fd1:1,fd2:1....max_num:100]
if ($platform == 4) {// 目前只有极光做了 隧道
$out_key = "aur_tunnel_online_" . $tokenall;
// 先验证
$have_fd = $redis->hExists($out_key, $fd);
if ($have_fd) {
$backbool = "true";
echo "有记录 {$fd}\n";
$re_mag = $this->send_msg($fd, $backbool); // 这里作为回复客户端
return "fd: {$fd} 触发的 Task {$task_id} 的 结果:{$re_mag}\n";
}
echo "该id 没有记录 {$fd}\n";
// 没有在里面 就 要重新搞了
$have_max = $redis->hExists($out_key, "max_num");
if (!$have_max) {
echo "没找到最大数 {$out_key}\n";
$tunnel_info = $this->set_max_num($prefix = "aur_", $tunnel_id, $token, $out_key);// 重置下
if (!empty($tunnel_info)) {
$max_num = $tunnel_info["online_max_num"];
$redis->hSet($out_key, "max_num", $max_num);
echo "设置后获取max_num：" . $redis->hGet($out_key, "max_num");
// $redis->expire($out_key,5*60);// 60s 从库里面校验
}else {// 这里 要 做下阻挡 由于 非法token 一直查询不到 ，每次过来查库 对 数据库造成压力
//bad_token 入库
$redis->set("aur_bad_token_" . $tokenall, NOW_TIME + $time_length);
$max_num = 0; //这里很重要 就是 当 token 不对 时 $max_num
}
}else {
$max_num = $redis->hGet($out_key, "max_num");
}
$num_now = $redis->hLen($out_key)- 1;// 里面多了一个 键max_num
echo " {$out_key}最大数：{$max_num},现在数：$num_now\n";
if (!empty($max_num)&& $max_num > $num_now) {
// 验证一个 并且放入 （有就算了）
$new_fd = $redis->hSet($out_key, $fd, $fd);
$map_up = $redis->hSet($this->all_fd_token_map, $fd, $out_key); // all_tunnel_online_map:[fd1:aurtoken,fd2:inttoken2,fd3:token2]
echo "new_fd {$fd} 入库 \n";
echo "{$fd}:{$out_key}map 入库结果:{$map_up}\n";
var_dump($new_fd);
$backbool = "true";
}
}else {// 如果有其他的 请在这里 做分支判断
}
$redis->close();
$msg = $this->send_msg($fd, $backbool); // 这里作为回复客户端
return "fd: {$fd} Task {$task_id}'s 结果{$msg}";
}
private function bad_token_check($bad_token_key)
{
$redis = new \Redis();
$redis->pconnect($this->redis_server, $this->redis_port);
if (!empty($this->redis_pwd)) {
$redis->auth($this->redis_pwd);
}
$expire = $redis->get($bad_token_key);
echo "bad 过期时间是：{$expire}\n";
if ($expire > NOW_TIME) {//被锁了
echo "{$bad_token_key}这token是个坏小子\n";
return false;
}
return true;
}
/**
* @param string $prefix
* @param $tunnel_id
* @param $token
* @param $out_key
* @return bool|mixed
*/
private function set_max_num($prefix = "aur_", $tunnel_id, $token, $out_key)
{
// 矫正用
echo "矫正ing....................................数据库查询\n";
$tunnel_info = Db::connect($this->db_config)->table($prefix . "tunnel_user_package")->where(['id' => $tunnel_id])->find();
if (!$tunnel_info) {
return false;
}
if (md6($tunnel_id . "lingjiang735" . $tunnel_info["salt"])!= $token) {
return false;
}
return $tunnel_info;
}
public function send_msg($fd, $msg)
{
$reminder = "向->{$fd} 发送-> {$msg}\n";
$this->serv->send($fd, $msg);
return $reminder;
}
public function onFinish($serv, $task_id, $data)
{
echo "Task {$task_id} over\n";
echo "Finish: {$data}\n";
}
public function onClose($serv, $fd, $from_id)
{
echo "1 Client {$fd} close connection\n";
// 这个端的唯一 链接 id
$redis = new \Redis();
$redis->pconnect($this->redis_server, $this->redis_port);
if (!empty($this->redis_pwd)) {
$redis->auth($this->redis_pwd);
}
$have_map = $redis->hExists($this->all_fd_token_map, $fd);
echo "2 {$fd}是否有map?:\n";
if ($have_map) {
$token_key = $redis->hGet($this->all_fd_token_map, $fd);
echo "3 {$fd}查询到token_key:{$token_key}\n";
//删除该token_key 下的
$re = $redis->hDel($token_key, $fd);
echo "4 {$fd}删除结果：$re\n";
echo "over\n";
}else {
echo "3 {$fd}没有查询到token_key\n";
}
$redis->close();
}
private function is_json($str)
{
return is_array(json_decode($str, true))&& !empty(json_decode($str));
}
/**
* 从数据库拿到
*/
private function set_config()
{
$m = Db::connect($this->db_config)->table('wt_config');
$r = $m->select();
foreach ($r as $k => $v) {
$r[$k]['name']= strtoupper($r[$k]['code']);
}
$r = array_column($r, 'value', 'code');
cache('config_cache', $r);
//取配置,赋值
config(cache('config_cache')); //添加配置
echo "设置缓存";
}
private function clean_all_tunnel_key()
{
$redis = new \Redis();
$redisserver = $this->redis_server;
$redisport = $this->redis_port;
$redispwd = $this->redis_pwd;
$redis->pconnect($redisserver, $redisport);
if (!empty($redispwd)) {
$redis->auth($redispwd);
}
echo "清理前各个token_list ：\n";
$infos = $redis->keys('aur_tunnel_online_*');
dump($infos);
$redis->delete($infos);
echo "清理前各个token_list ：\n";
$infos = $redis->keys('aur_tunnel_online_*');
dump($infos);
echo "清理前map：\n";
$infos = $redis->keys($this->all_fd_token_map);
dump($infos);
$redis->delete($infos);
echo "清理前各个token_list and map：\n";
$infos = $redis->keys($this->all_fd_token_map);
dump($infos);
}
}
