tcjlj/Xboard
1
0
Fork 0
mirror of https://github.com/cedar2025/Xboard.git synced 2025-01-22 18:48:14 -05:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

fix: 优化流量消费队列，防止记录长时间被锁住

Browse Source
This commit is contained in:
xboard 2023-12-04 23:16:18 +08:00
parent 66ab4a4a8e
commit 64cc2d79da
2 changed files with 24 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
app/Jobs/TrafficFetchJob.php
View File

@ -20,7 +20,7 @@ class TrafficFetchJob implements ShouldQueue
protected $childServer; protected $childServer;
protected $protocol; protected $protocol;
protected $timestamp; protected $timestamp;
public $tries = 3; public $tries = 1;
public $timeout = 10; public $timeout = 10;
/** /**
@ -28,46 +28,37 @@ class TrafficFetchJob implements ShouldQueue
* *
* @return void * @return void
*/ */
public function __construct(array $server, array $data, $protocol, int $timestamp, $nodeIp = null) public function __construct(array $server, array $data, $protocol, int $timestamp, $childServer = null)
{ {
$this->onQueue('traffic_fetch'); $this->onQueue('traffic_fetch');
$this->server = $server; $this->server = $server;
$this->data = $data; $this->data = $data;
$this->protocol = $protocol; $this->protocol = $protocol;
$this->timestamp = $timestamp; $this->timestamp = $timestamp;
// 获取子节点 $this->childServer = $childServer;
$serverService = new ServerService();
$this->childServer = ($this->server['parent_id'] == null && !blank($nodeIp)) ? $serverService->getChildServer($this->server['id'], $this->protocol, $nodeIp) : null;
} }
/**
* Execute the job.
*
* @return void
*/
public function handle(): void public function handle(): void
{ {
\DB::transaction(function () { if ($this->attempts() === 1) {
if ($this->attempts() === 1){ $statService = new StatisticalService();
$statService = new StatisticalService(); $statService->setStartAt($this->timestamp);
$statService->setStartAt($this->timestamp); $statService->setUserStats();
$statService->setUserStats(); $statService->setServerStats();
$statService->setServerStats(); }
} // 获取子节点\
$targetServer = $this->childServer ?? $this->server;
// 获取子节点\ foreach ($this->data as $uid => $v) {
\DB::transaction(function () use ($uid, $v, $targetServer, $statService) {
$targetServer = $this->childServer ?? $this->server;
foreach ($this->data as $uid => $v) {
$u = $v[0]; $u = $v[0];
$d = $v[1]; $d = $v[1];
$user = \DB::table('v2_user')->lockForUpdate()->where('id', $uid)->first(); $user = \DB::table('v2_user')->lockForUpdate()->where('id', $uid)->first();
if (!$user) { if (!$user) {
continue; return;
} }
if ($this->attempts() === 1){ // 写缓存 if ($this->attempts() === 1) { // 写缓存
$statService->statUser($targetServer['rate'], $uid, $u, $d); //如果存在子节点则使用子节点的倍率 $statService->statUser($targetServer['rate'], $uid, $u, $d); //如果存在子节点则使用子节点的倍率
if(!blank($this->childServer)){ //如果存在子节点，则给子节点计算流量 if (!blank($this->childServer)) { //如果存在子节点，则给子节点计算流量
$statService->statServer($this->childServer['id'], $this->protocol, $u, $d); $statService->statServer($this->childServer['id'], $this->protocol, $u, $d);
} }
$statService->statServer($this->server['id'], $this->protocol, $u, $d); $statService->statServer($this->server['id'], $this->protocol, $u, $d);
@ -82,7 +73,7 @@ class TrafficFetchJob implements ShouldQueue
'u' => $newU, 'u' => $newU,
'd' => $newD, 'd' => $newD,
]); ]);
} }, 3);
}, 3); }
} }
} }

8
app/Services/UserService.php
View File

@ -170,9 +170,13 @@ class UserService
public function trafficFetch(array $server, string $protocol, array $data, string $nodeIp = null) public function trafficFetch(array $server, string $protocol, array $data, string $nodeIp = null)
{ {
// 获取子节点
$childServer = ($server['parent_id'] == null && !blank($nodeIp))
? (new ServerService())->getChildServer($server['id'], $protocol, $nodeIp)
: null;
$timestamp = strtotime(date('Y-m-d')); $timestamp = strtotime(date('Y-m-d'));
collect($data)->chunk(1000)->each(function($chunk) use ($timestamp,$server,$protocol, $nodeIp){ collect($data)->chunk(1000)->each(function($chunk) use ($timestamp,$server,$protocol, $childServer){
TrafficFetchJob::dispatch($server, $chunk->toArray(), $protocol, $timestamp, $nodeIp); TrafficFetchJob::dispatch($server, $chunk->toArray(), $protocol, $timestamp, $childServer);
}); });
} }
} }