perf: 优化用户流量消费队列（上万用户流量信息数秒即可处理完成

app/Http/Controllers/V1/Server/UniProxyController.php

@@ -89,12 +89,8 @@ class UniProxyController extends Controller
         $online_user = $onlineCollection->sum('online_user');
         Cache::put(CacheKey::get('SERVER_' . strtoupper($this->nodeType) . '_ONLINE_USER', $this->nodeInfo->id), $online_user, 3600);
         Cache::put(CacheKey::get('SERVER_' . strtoupper($this->nodeType) . '_LAST_PUSH_AT', $this->nodeInfo->id), time(), 3600);
-
-        // 查询是否存在子节点
-        $childServer = null;
-        if ($this->nodeInfo->parent_id == null) $childServer = $this->serverService->getChildServer($this->nodeId, $this->nodeType, $ip);
         $userService = new UserService();
-        $userService->trafficFetch($this->nodeInfo->toArray(), $this->nodeType, $data, $childServer ? $childServer->toArray() : null);
+        $userService->trafficFetch($this->nodeInfo->toArray(), $this->nodeType, $data , $ip);
 
         return response([
             'data' => true

app/Jobs/TrafficFetchJob.php

@@ -4,6 +4,8 @@ namespace App\Jobs;
 
 use App\Models\User;
 use App\Services\MailService;
+use App\Services\ServerService;
+use App\Services\StatisticalService;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
@@ -13,12 +15,11 @@ use Illuminate\Queue\SerializesModels;
 class TrafficFetchJob implements ShouldQueue
 {
     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
-    protected $u;
+    protected $data;
-    protected $d;
-    protected $userId;
     protected $server;
+    protected $childServer;
     protected $protocol;
-
+    protected $timestamp;
     public $tries = 3;
     public $timeout = 10;
 
@@ -27,14 +28,16 @@ class TrafficFetchJob implements ShouldQueue
      *
      * @return void
      */
-    public function __construct($u, $d, $userId, array $server, $protocol)
+    public function __construct(array $server, array $data, $protocol, int $timestamp, $nodeIp = null)
     {
         $this->onQueue('traffic_fetch');
-        $this->u = $u;
-        $this->d = $d;
-        $this->userId = $userId;
         $this->server = $server;
+        $this->data = $data;
         $this->protocol = $protocol;
+        $this->timestamp = $timestamp;
+        // 获取子节点
+        $serverService = new ServerService();
+        $this->childServer = ($this->server['parent_id'] == null && !blank($nodeIp)) ?  $serverService->getChildServer($this->server['id'], $this->protocol, $nodeIp) : null;
     }
 
     /**
@@ -45,29 +48,40 @@ class TrafficFetchJob implements ShouldQueue
     public function handle(): void
     {
         \DB::transaction(function () {
-            $user = \DB::table('v2_user')->lockForUpdate()->where('id', $this->userId)->first();
+            if ($this->attempts() === 1){
-        
+                $statService = new StatisticalService();
-            if (!$user) {
+                $statService->setStartAt($this->timestamp);
-                return;
+                $statService->setUserStats();
+                $statService->setServerStats();
             }
-        
+            
-            $newTime = time();
+            // 获取子节点\
-            $newU = $user->u + ($this->u * $this->server['rate']);
+            
-            $newD = $user->d + ($this->d * $this->server['rate']);
+            $targetServer = $this->childServer ?? $this->server;
-        
+            foreach ($this->data as $uid => $v) {
-            $updatedRows = \DB::table('v2_user')
+                $u = $v[0];
-                ->where('id', $this->userId)
+                $d = $v[1];
-                ->update([
+                $user = \DB::table('v2_user')->lockForUpdate()->where('id', $uid)->first();
-                    't' => $newTime,
+                if (!$user) {
-                    'u' => $newU,
+                    continue;
-                    'd' => $newD,
+                }
-                ]);
+                if ($this->attempts() === 1){ // 写缓存
-        
+                    $statService->statUser($targetServer['rate'], $uid, $u, $d); //如果存在子节点则使用子节点的倍率
-            if (!$updatedRows) {
+                    if(!blank($this->childServer)){ //如果存在子节点，则给子节点计算流量
-                info("流量更新失败\n未记录用户ID:{$this->userId}\n未记录上行:{$this->u}\n未记录下行:{$this->d}");
+                        $statService->statServer($this->childServer['id'], $this->protocol, $u, $d);
-                $this->fail();
+                    }
-            } else {
+                    $statService->statServer($this->server['id'], $this->protocol, $u, $d);
-                
+                }
+                $newTime = time();
+                $newU = $user->u + ($v[0] * $targetServer['rate']);
+                $newD = $user->d + ($v[1] * $targetServer['rate']);
+                \DB::table('v2_user')
+                    ->where('id', $uid)
+                    ->update([
+                        't' => $newTime,
+                        'u' => $newU,
+                        'd' => $newD,
+                    ]);
             }
         }, 3);
     }

app/Services/ServerService.php

@@ -363,7 +363,7 @@ class ServerService
                         ->where('ips',"like", "%\"$nodeIp\"%")
                         ->first();
             default:
-                return false;
+                return null;
         }
     }
 }

app/Services/UserService.php

@@ -168,26 +168,11 @@ class UserService
         return true;
     }
 
-    public function trafficFetch(array $server, string $protocol, array $data, array $childServer = null)
+    public function trafficFetch(array $server, string $protocol, array $data, string $nodeIp = null)
     {
-        $statService = new StatisticalService();
+        $timestamp = strtotime(date('Y-m-d'));
-        $statService->setStartAt(strtotime(date('Y-m-d')));
+        collect($data)->chunk(1000)->each(function($chunk) use ($timestamp,$server,$protocol, $nodeIp){
-        $statService->setUserStats();
+            TrafficFetchJob::dispatch($server, $chunk->toArray(), $protocol, $timestamp, $nodeIp);
-        $statService->setServerStats();
+        });
-        foreach (array_keys($data) as $userId) {
-            $u = $data[$userId][0];
-            $d = $data[$userId][1];
-            // 如果存在子节点则使用过子节点的倍率进行进行流量计算，该计算方式依赖服务器IP地址
-            if(!blank($childServer)){
-                TrafficFetchJob::dispatch($u, $d, $userId, $childServer, $protocol);
-                $statService->statUser($childServer['rate'], $userId, $u, $d);
-                $statService->statServer($childServer['id'], $protocol, $u, $d);
-            }else{
-                TrafficFetchJob::dispatch($u, $d, $userId, $server, $protocol);
-                $statService->statUser($server['rate'], $userId, $u, $d);
-            }
-            $statService->statServer($server['id'], $protocol, $u, $d);
-
-        }
     }
 }