测试工具 http://www.blue-zero.com/WebSocket/
2018年8月6日17:28:24
<?php namespace App\Console\Commands; use Illuminate\Console\Command;
use Workerman\Worker;
use App\Work\ChatroomWork; class Chatroom extends Command { protected $taskserver;
/*
* 操作参数
* 注意只能在
* start 启动
* stop 停止
* relaod 只能重启逻辑代码，核心workerman_init无法重启，注意看官方文档
* status 查看状态
* connections 查看连接状态（需要Workerman版本>=3.5.0）
*
* 库 composer require workerman/workerman
*/
protected $action = array('start', 'stop', 'reload', 'status', 'connections'); /**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'Chatroom {action}'; /**
* The console command description.
*
* @var string
*/
protected $description = 'Chatroom'; /**
* Create a new command instance.
*
* @return void
*/
public function __construct() {
parent::__construct();
} /**
* Execute the console command.
*
* @return mixed
*
* 注意上线提供的方法
*
*/
public function handle() {
$action = $this->argument('action'); if (!in_array($action, $this->action)) {
$this->error('Error Action');
exit;
}
//初始化workerman
ChatroomWork::workerman_init($action);
} }
<?php namespace App\Work; use App\Work\BaseWork as Base;
use Illuminate\Support\Facades\DB;
use App\Work\CommonWork;
use Workerman\Worker;
use Workerman\Lib\Timer;
use App\Models\OperationLog;
use App\Models\Users; class ChatroomWork extends Base { //全局的总连接数
static $connection_count = 0;
//单个房间最大的连接数
static $room_max_numbers;
//房间所有的用户连接数ID集合，判断发送给那些用户，绑定用户和connection_id
static $room_connection_array = []; public static function workerman_init($action = null) {
global $argv; $argv[0] = 'workerman:websocket';
$argv[1] = $action;
// $argv[2] = '-d';
// 心跳
define('HEARTBEAT_TIME', 30);
//初始化
$worker = new Worker("websocket://172.17.1.247:9090");
$worker->name = 'Chatroom';
//linux 用户线上是www
// $worker->user = 'www';
//守护模式信息输出文件地址
// $worker->stdoutFile = "./workerman.log";
//工作进程总数 测试环境4个
$worker->count = 4;
//正式环境
// $ws->count = 10;
//建立链接 处理逻辑
$worker->onConnect = function($connection) {
// 有新的客户端连接时，连接数+1
self::$connection_count++;
self::onConnect($connection);
};
//接受消息 处理逻辑
$worker->onMessage = function($connection, $data) { self::onMessage($connection, $data);
};
//关闭链接 处理逻辑
$worker->onClose = function($connection) {
// 客户端关闭时，连接数-1
self::$connection_count--;
self::onClose($connection);
}; // 进程启动后设置一个30秒运行一次的定时器
$worker->onWorkerStart = function($worker) {
// Timer::add(1, function()use($worker) {
// $time_now = time();
// foreach ($worker->connections as $connection) {
//
// }
// });
};
// 开始
Worker::runAll();
} //建立链接 处理逻辑
public static function onConnect($connection) {
//主动心跳ping测试60秒一次
// Timer::add(HEARTBEAT_TIME, function() use($connection) {
// $connection->send(json_encode(['code' => 200, 'msg' => '服务存活', 'data' => [], 'connections' => self::$connection_count]));
// });
} //接受消息 处理逻辑
public static function onMessage($connection, $data) {
//解析数据，非合法的json数据不处理 if (!empty($data)) {
if (is_json($data)) {
$data = json_decode($data, true); if ($data['action_type'] == 'ping') {
// 客户端回应服务端的心跳
$connection->send(json_encode(['code' => 200, 'msg' => '服务存活', 'data' => [], 'connections' => self::$connection_count]));
} elseif ($data['action_type'] == 'login') {
//用户登录
if ($data['is_login'] == 1) {
//匿名登录
self::$room_connection_array[$data['room_id']][$connection->id]['user_name'] = '匿名用户' . $connection->id; $connection->send(json_encode(['code' => 200, 'msg' => '匿名登录成功', 'data' => self::$room_connection_array, 'connections' => self::$connection_count]));
} elseif ($data['is_login'] == 2) {
//已登录
$Users = Users::where('id', $data['user_id'])->first();
if (empty($Users)) {
$connection->send(json_encode(['code' => 201, 'msg' => '用户ID无效或者错误', 'data' => [], 'connections' => self::$connection_count]));
} else {
$Users = $Users->toArray();
self::$room_connection_array[$data['room_id']][$connection->id]['user_name'] = empty($Users['realname']) ? $data['user_id'] : $Users['realname']; $connection->send(json_encode(['code' => 200, 'msg' => '登录成功', 'data' => self::$room_connection_array, 'connections' => self::$connection_count]));
}
} else {
$connection->send(json_encode(['code' => 201, 'msg' => '登录类型数据错误', 'data' => [], 'connections' => self::$connection_count]));
}
} elseif ($data['action_type'] == 'broadcast_to_all') {
//只发给房间的所有的人，除去自己
foreach ($connection->worker->connections as $con) {
foreach (self::$room_connection_array[$data['room_id']] as $k => $v) {
// p($con->id);
// p($k);
if ($k == $con->id && $con->id != $connection->id) {
$con->send($data['message']);
}
}
}
} elseif ($data['action_type'] == 'broadcast_to_one') { } else {
$connection->send(json_encode(['code' => 201, 'msg' => 'action_type类型错误', 'data' => [], 'connections' => self::$connection_count]));
}
}
} else {
$connection->send(json_encode(['code' => 201, 'msg' => '数据请求为空', 'data' => [], 'connections' => self::$connection_count]));
}
} //关闭链接 处理逻辑
public static function onClose($connection) {
//检索$connection->id 是否有在self::$room_connection_array中，有的话就剔除 foreach (self::$room_connection_array as $k => $v) {
foreach ($v as $kk => $vv) {
if ($kk == $connection->id) {
unset(self::$room_connection_array[$k][$connection->id]);
}
}
}
// p(self::$room_connection_array['100']);
} }
{"action_type":"login","is_login":1,"room_id":101} //登录
{"action_type":"broadcast_to_all","room_id":100,"message":"111"} //发送消息
为什么不用switch代码会看起来更清晰，因为有bug，对字符串匹配的不好
注意：
1，$connection就是当前的连接数据
2，因为未根据work id做房间划分，不知道在超出单个work连接时候会不会出问题
3，可以根据实际压力去划分一个work的最大连接数，这里是简单的测试demo所以未做具体的细节划分
4，这里应该结合session来做处理数据，但是我只是根据发送数据来区别用户，你可以在登录的发送数据的时候根据session处理数据，看需要，必须要返回用户列表，就直接把room id下面的所有用户名返回就OK
5，workerman用起来其实还是比较简单的，但是我这种结合laravel的整合是有问题，比如现在我有一个消息推送，一个聊天室就没办法放在一起使用，必须用别的办法，如果是单功能就比较容易，直接结合，我有一个骚办法就是直接
复制artisan入口文件，直接增加新入口artisan1，经过测试完全没有问题，但是其实不是很好的解决方案，如果要用就先这么上吧 如果你觉得麻烦可以是gateway做比较简单
2019年7月12日09:43:56
注意：上面是临时测试代码。业务代码使用try catch处理异常和错误