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

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

1.首先删除vordor文件夹

2.安装扩展,下载workerman

1.  
   //安装扩展
2.  
   yum install php-process
3.  
   //下载workerman
4.  
   composer require topthink/think-worker

3.在application创建server.php

代码:

1.  
   <?php
2.  
   define('APP_PATH', __DIR__ . '/../application/');
3.  
   //定义监听控制器
4.  
   define('BIND_MODULE','push/Workerman');
5.  
    
6.  
    
7.  
    
8.  
   // 加载框架引导文件
9.  
    
10.  
    require __DIR__ . '/../thinkphp/start.php';

4.在application目录创建push目录

application/push/controller/Worker.php

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

代码:

1.  
   <?php
2.  
   namespace app\push\controller;
3.  
   use think\worker\Server;
4.  
   use Workerman\Lib\Timer;
5.  
   use think\Db;
6.  
    
7.  
   class Worker extends Server{
8.  
   protected $socket = 'http://0.0.0.0:2348'; //linux服务器端口
9.  
   protected static $heartbeat_time=55;
10.  
     
11.  
    /**
12.  
    * 收到信息
13.  
    * @param $connection
14.  
    * @param $data
15.  
    */
16.  
    public function onMessage($connection, $data)
17.  
    {
18.  
    if($data=="ping"&&$data==0){
19.  
    }else{
20.  
    //接收的参数
21.  
    }
22.  
    $connection->send("ping");
23.  
    $connection->lastMessageTime=time();
24.  
    }
25.  
     
26.  
     
27.  
    /**
28.  
    * 每个进程启动
29.  
    * @param $worker
30.  
    */
31.  
    public function onWorkerStart($worker){
32.  
    //查看是否有新的充值或提现订单，有就推送给所有用户
33.  
    Timer::add(3, function()use($worker){
34.  
     
35.  
    $time_now=time();
36.  
    $hasNewDepositOrder = Db::name('worker')->where('is_push',0)->order('id desc')->count('id');
37.  
    // $system_listener = Db::name('worker')->cache(true)->order('id desc')->select();
38.  
     
39.  
    if($hasNewDepositOrder){
40.  
    $depositOrderInfo = Db::name('worker')->where('is_push',0)->order('id desc')->find();
41.  
    $data = ['creatTime'=>date('Y-m-d H:i:s'),'name'=>$depositOrderInfo['name'],'tel'=>$depositOrderInfo['tel']];
42.  
    foreach($worker->connections as $connection) {
43.  
    if(empty($connection->lastMessageTime)){
44.  
    $connection->lastMessageTime = $time_now;
45.  
    }
46.  
     
47.  
    if($time_now-$connection->lastMessageTime > self::$heartbeat_time){
48.  
     
49.  
    $connection->close();
50.  
    }
51.  
     
52.  
    $connection->send(json_encode($data));
53.  
    }
54.  
     
55.  
    Db::name('worker')->where('id',$depositOrderInfo['id'])->update(['is_push'=>1]);
56.  
    }else{
57.  
    foreach($worker->connections as $connection) {
58.  
    if(empty($connection->lastMessageTime)){
59.  
    $connection->lastMessageTime = $time_now;
60.  
    continue;
61.  
    }
62.  
     
63.  
    if($time_now-$connection->lastMessageTime > self::$heartbeat_time){ //连接超时
64.  
     
65.  
    $connection->close();
66.  
    }
67.  
    }
68.  
    }
69.  
    });
70.  
     
71.  
     
72.  
    }
73.  
     
74.  
    }

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

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

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

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

整体代码如下:

1.  
   <?php
2.  
   // ----------------------------------------------------------------------
3.  
   // | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
4.  
   // ----------------------------------------------------------------------
5.  
   // | Copyright (c) 2006-2014 http://thinkphp.cn All rights reserved.
6.  
   // ----------------------------------------------------------------------
7.  
   // | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
8.  
   // ----------------------------------------------------------------------
9.  
   // | Author: liu21st <liu21st@gmail.com>
10.  
    // ----------------------------------------------------------------------
11.  
     
12.  
    namespace think\worker;
13.  
     
14.  
    use Workerman\Worker;
15.  
     
16.  
    /**
17.  
    * Worker控制器扩展类
18.  
    */
19.  
    abstract class Server
20.  
    {
21.  
    protected $worker;
22.  
    protected $socket = '';
23.  
    protected $protocol = 'http';
24.  
    protected $host = '0.0.0.0';
25.  
    protected $port = '2346';
26.  
    protected $processes = 4;
27.  
     
28.  
    /**
29.  
    * 架构函数
30.  
    * @access public
31.  
    */
32.  
    public function __construct()
33.  
    {
34.  
    // 实例化 Websocket 服务
35.  
     
36.  
    $this->worker = new Worker($this->socket);//this->socket
37.  
    // 设置进程数
38.  
    $this->worker->count = $this->processes;
39.  
    // 初始化
40.  
    $this->init();
41.  
     
42.  
    // 设置回调
43.  
    foreach (['onWorkerStart', 'onConnect', 'onMessage', 'onClose', 'onError', 'onBufferFull', 'onBufferDrain', 'onWorkerStop', 'onWorkerReload'] as $event) {
44.  
    if (method_exists($this, $event)) {
45.  
    $this->worker->$event = [$this, $event];
46.  
    }
47.  
    }
48.  
    // Run worker
49.  
    Worker::runAll();
50.  
    }
51.  
     
52.  
    protected function init()
53.  
    {
54.  
    }
55.  
     
56.  
    }

这篇好文章是转载于：学新通技术网