Thinkphp5+FastAdmin配置workerman消息推送(多线程)

环境：linux系统、fastadmin（tp5内核）、php7.4.3，workerman1.0

1.首先删除vordor文件夹

2.安装扩展,下载workerman

//安装扩展
yum install php-process
//下载workerman
composer require topthink/think-worker

3.在application创建server.php

代码:

<?php
define('APP_PATH', __DIR__ . '/../application/');
 //定义监听控制器
define('BIND_MODULE','push/Workerman');
 
// 加载框架引导文件
 
require __DIR__ . '/../thinkphp/start.php';

4.在application目录创建push目录

application/push/controller/Worker.php

注:如果想监听多个端口,只需要在application/push/controller中再创建一个文件,把端口改一下即可,另外在application创建再创建一个server.php

代码:

<?php
namespace app\push\controller; 
use think\worker\Server; 
use Workerman\Lib\Timer;
use think\Db;
 
class Worker extends Server{    
    protected $socket = 'http://0.0.0.0:2348'; //linux服务器端口
    protected static $heartbeat_time=55;
    
      /**
     * 收到信息
     * @param $connection
     * @param $data
     */
    public function onMessage($connection, $data)
    {
        if($data=="ping"&&$data==0){
        }else{
        //接收的参数
        }
        $connection->send("ping");
        $connection->lastMessageTime=time();
    }
    
   
    /**
     * 每个进程启动
     * @param $worker
     */
    public function onWorkerStart($worker){
    //查看是否有新的充值或提现订单，有就推送给所有用户
        Timer::add(3, function()use($worker){
 
            $time_now=time();
            $hasNewDepositOrder   =   Db::name('worker')->where('is_push',0)->order('id desc')->count('id');
            // $system_listener    =   Db::name('worker')->cache(true)->order('id desc')->select();
 
            if($hasNewDepositOrder){
                $depositOrderInfo   =   Db::name('worker')->where('is_push',0)->order('id desc')->find();
                $data   =   ['creatTime'=>date('Y-m-d H:i:s'),'name'=>$depositOrderInfo['name'],'tel'=>$depositOrderInfo['tel']];
                foreach($worker->connections as $connection) {
                    if(empty($connection->lastMessageTime)){
                        $connection->lastMessageTime =   $time_now;
                    }
 
                    if($time_now-$connection->lastMessageTime > self::$heartbeat_time){
                        
                        $connection->close();
                    }
 
                    $connection->send(json_encode($data));
                }
 
              Db::name('worker')->where('id',$depositOrderInfo['id'])->update(['is_push'=>1]);
            }else{
                foreach($worker->connections as $connection) {
                    if(empty($connection->lastMessageTime)){
                        $connection->lastMessageTime = $time_now;
                        continue;
                    }
 
                    if($time_now-$connection->lastMessageTime > self::$heartbeat_time){      //连接超时    $connection->close();
                    }
                }
            }
        });
 
 
    }
 
}

5.找到/vendor/topthink/think-worker/src里面的Server.php

6.使用命令进入到application目录中,执行命令:php server.php start

注:如果想要监听多个端口:需要找到$this->worker = new Worker();

改成$this->worker = new Worker($this->socket);

整体代码如下:

<?php
// +----------------------------------------------------------------------
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
// +----------------------------------------------------------------------
// | Copyright (c) 2006-2014 http://thinkphp.cn All rights reserved.
// +----------------------------------------------------------------------
// | Licensed ( https://baitexiaoyuan.oss-cn-zhangjiakou.aliyuncs.com/updatecrm/php/20230922/LICENSE-2.0 )
// +----------------------------------------------------------------------
// | Author: liu21st <liu21st@gmail.com>
// +----------------------------------------------------------------------namespace think\worker;use Workerman\Worker;/**
 * Worker控制器扩展类
 */
abstract class Server
{
    protected $worker;
    protected $socket    = '';
    protected $protocol  = 'http';
    protected $host      = '0.0.0.0';
    protected $port      = '2346';
    protected $processes = 4;    /**
     * 架构函数
     * @access public
     */
    public function __construct()
    {
        // 实例化 Websocket 服务
       
        $this->worker = new Worker($this->socket);//this->socket
        // 设置进程数
        $this->worker->count = $this->processes;
        // 初始化
        $this->init();        // 设置回调
        foreach (['onWorkerStart', 'onConnect', 'onMessage', 'onClose', 'onError', 'onBufferFull', 'onBufferDrain', 'onWorkerStop', 'onWorkerReload'] as $event) {
            if (method_exists($this, $event)) {
                $this->worker->$event = [$this, $event];
            }
        }
        // Run worker
        Worker::runAll();
    }    protected function init()
    {
    }}