Skip to content

Commit 56e9717

Browse files
fun90fun90
committed
fix(guard): 重试在线状态并发写入
1 parent 5021204 commit 56e9717

3 files changed

Lines changed: 106 additions & 14 deletions

File tree

‎src/main/java/com/fun90/airopscat/repository/AccountOnlineIpRepository.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ public int deleteExpiredRecordsBatch(LocalDateTime expireTime, int batchSize) {
185185
* 使用 MySQL 原生 UPSERT 原子更新在线状态。
186186
* 如果距离上次续期超过离线阈值，则重新开始计算本次在线会话时间。
187187
*/
188-
@Transactional
188+
@Transactional(Transactional.TxType.REQUIRES_NEW)
189189
public void upsertOnlineStatus(String accountNo,
190190
String clientIp,
191191
String connectionId,

‎src/main/java/com/fun90/airopscat/service/AccountOnlineIpService.java‎

Lines changed: 77 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,12 @@
1313
import com.fun90.airopscat.repository.UserRepository;
1414
import jakarta.enterprise.context.ApplicationScoped;
1515
import jakarta.inject.Inject;
16+
import jakarta.persistence.LockTimeoutException;
17+
import jakarta.persistence.PessimisticLockException;
1618
import jakarta.transaction.Transactional;
1719
import lombok.extern.slf4j.Slf4j;
1820

21+
import java.sql.SQLTransientException;
1922
import java.time.LocalDateTime;
2023
import java.time.OffsetDateTime;
2124
import java.time.ZoneId;
@@ -32,6 +35,8 @@
3235
@Slf4j
3336
public class AccountOnlineIpService {
3437
private static final int DEFAULT_CLEANUP_BATCH_SIZE = 1000;
38+
private static final int UPSERT_MAX_ATTEMPTS = 3;
39+
private static final long UPSERT_RETRY_BACKOFF_MILLIS = 50L;
3540

3641
private final AccountOnlineIpRepository accountOnlineIpRepository;
3742
private final AccountRepository accountRepository;
@@ -139,7 +144,6 @@ public long deleteByAccountNo(String accountNo) {
139144
* @param onlineAccountIps guard agent 上报的账号在线 IP 聚合记录
140145
* @return 本轮成功 upsert 的记录数
141146
*/
142-
@Transactional
143147
public int refreshFromGuardAccountIps(String nodeIp, List<GuardOnlineAccountIpReport> onlineAccountIps) {
144148
if (nodeIp == null || nodeIp.isBlank() || onlineAccountIps == null) {
145149
return 0;
@@ -196,13 +200,10 @@ public int refreshFromGuardAccountIps(String nodeIp, List<GuardOnlineAccountIpRe
196200
continue;
197201
}
198202
String connectionId = buildGuardAccountIpConnectionId(accountNo, clientIp, normalizedNodeIp, nodeTag);
199-
try {
200-
accountOnlineIpRepository.upsertOnlineStatus(accountNo, clientIp, connectionId, normalizedNodeIp,
201-
node == null ? null : node.getId(), nodeTag, now, now, now, now, offlineThreshold);
203+
if (upsertOnlineStatusWithRetry(accountNo, clientIp, connectionId, normalizedNodeIp,
204+
node == null ? null : node.getId(), nodeTag, now, now, now, now, offlineThreshold,
205+
"refreshFromGuardAccountIps")) {
202206
count++;
203-
} catch (Exception e) {
204-
log.error("refreshFromGuardAccountIps upsert 失败: accountNo={}, clientIp={}, connectionId={}, nodeIp={}",
205-
accountNo, clientIp, connectionId, normalizedNodeIp, e);
206207
}
207208
}
208209
}
@@ -232,13 +233,10 @@ private int refreshFromGuardConnectionRefs(List<GuardOnlineConnectionRefReport>
232233
continue;
233234
}
234235
LocalDateTime sessionStartTime = resolveConnectionStartTime(connection.getStart(), now);
235-
try {
236-
accountOnlineIpRepository.upsertOnlineStatus(accountNo, clientIp, connectionId, nodeIp,
237-
node == null ? null : node.getId(), nodeTag, now, sessionStartTime, now, now, offlineThreshold);
236+
if (upsertOnlineStatusWithRetry(accountNo, clientIp, connectionId, nodeIp,
237+
node == null ? null : node.getId(), nodeTag, now, sessionStartTime, now, now, offlineThreshold,
238+
"refreshFromGuardConnectionRefs")) {
238239
count++;
239-
} catch (Exception e) {
240-
log.error("refreshFromGuardConnectionRefs upsert 失败: accountNo={}, clientIp={}, connectionId={}, nodeIp={}",
241-
accountNo, clientIp, connectionId, nodeIp, e);
242240
}
243241
}
244242
return count;
@@ -256,6 +254,72 @@ public List<AccountOnlineIpDto> getOnlineRecordsByAccountNos(List<String> accoun
256254
return convertToDtoList(records);
257255
}
258256

257+
private boolean upsertOnlineStatusWithRetry(String accountNo,
258+
String clientIp,
259+
String connectionId,
260+
String nodeIp,
261+
Long nodeId,
262+
String nodeTag,
263+
LocalDateTime lastOnlineTime,
264+
LocalDateTime sessionStartTime,
265+
LocalDateTime createTime,
266+
LocalDateTime updateTime,
267+
LocalDateTime offlineThresholdTime,
268+
String source) {
269+
for (int attempt = 1; attempt <= UPSERT_MAX_ATTEMPTS; attempt++) {
270+
try {
271+
accountOnlineIpRepository.upsertOnlineStatus(accountNo, clientIp, connectionId, nodeIp,
272+
nodeId, nodeTag, lastOnlineTime, sessionStartTime, createTime, updateTime, offlineThresholdTime);
273+
return true;
274+
} catch (Exception e) {
275+
if (isRetryableUpsertFailure(e) && attempt < UPSERT_MAX_ATTEMPTS) {
276+
log.warn("{} upsert 遇到并发锁冲突，准备重试: accountNo={}, clientIp={}, connectionId={}, nodeIp={}, attempt={}/{}, error={}",
277+
source, accountNo, clientIp, connectionId, nodeIp, attempt, UPSERT_MAX_ATTEMPTS, e.getMessage());
278+
sleepBeforeRetry(attempt);
279+
continue;
280+
}
281+
log.error("{} upsert 失败: accountNo={}, clientIp={}, connectionId={}, nodeIp={}, attempt={}/{}",
282+
source, accountNo, clientIp, connectionId, nodeIp, attempt, UPSERT_MAX_ATTEMPTS, e);
283+
return false;
284+
}
285+
}
286+
return false;
287+
}
288+
289+
private boolean isRetryableUpsertFailure(Throwable throwable) {
290+
Throwable current = throwable;
291+
while (current != null) {
292+
if (current instanceof PessimisticLockException
293+
|| current instanceof LockTimeoutException
294+
|| current instanceof SQLTransientException) {
295+
return true;
296+
}
297+
String simpleName = current.getClass().getSimpleName();
298+
if (simpleName.contains("LockAcquisition") || simpleName.contains("TransactionRollback")) {
299+
return true;
300+
}
301+
String message = current.getMessage();
302+
if (message != null) {
303+
String lowerMessage = message.toLowerCase();
304+
if (lowerMessage.contains("deadlock found")
305+
|| lowerMessage.contains("try restarting transaction")
306+
|| lowerMessage.contains("lock wait timeout")) {
307+
return true;
308+
}
309+
}
310+
current = current.getCause();
311+
}
312+
return false;
313+
}
314+
315+
private void sleepBeforeRetry(int attempt) {
316+
try {
317+
Thread.sleep(UPSERT_RETRY_BACKOFF_MILLIS * attempt);
318+
} catch (InterruptedException e) {
319+
Thread.currentThread().interrupt();
320+
}
321+
}
322+
259323
/**
260324
* 清理超过指定小时数的历史在线记录
261325
*/

‎src/test/java/com/fun90/airopscat/service/AccountOnlineIpServiceGuardTest.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import com.fun90.airopscat.repository.AccountRepository;
99
import com.fun90.airopscat.repository.NodeRepository;
1010
import com.fun90.airopscat.repository.UserRepository;
11+
import jakarta.persistence.PessimisticLockException;
1112
import org.junit.jupiter.api.Test;
1213

1314
import java.time.LocalDateTime;
@@ -65,6 +66,26 @@ void shouldUseConnectionRefsWhenProvided() {
6566
assertEquals(LocalDateTime.of(2026, 7, 14, 10, 0), record.getSessionStartTime());
6667
}
6768

69+
@Test
70+
void shouldRetryGuardOnlineUpsertWhenDeadlockHappens() {
71+
FakeAccountOnlineIpRepository onlineIpRepository = new FakeAccountOnlineIpRepository();
72+
onlineIpRepository.failuresBeforeSuccess = 1;
73+
AccountOnlineIpService service = new AccountOnlineIpService(
74+
onlineIpRepository,
75+
new FakeAccountRepository(),
76+
new FakeNodeRepository(),
77+
new UserRepository(),
78+
new FakeSystemConfigService());
79+
80+
GuardOnlineAccountIpReport report = accountIpReport("acct-001", "node_7", List.of("203.0.113.10"));
81+
82+
int count = service.refreshFromGuardAccountIps("192.0.2.10", List.of(report));
83+
84+
assertEquals(1, count);
85+
assertEquals(2, onlineIpRepository.attempts);
86+
assertEquals(1, onlineIpRepository.records.size());
87+
}
88+
6889
private static GuardOnlineAccountIpReport accountIpReport(String accountNo,
6990
String nodeTag,
7091
List<String> clientIps) {
@@ -87,6 +108,8 @@ private static GuardOnlineConnectionRefReport connectionRef(String clientIp,
87108

88109
static class FakeAccountOnlineIpRepository extends AccountOnlineIpRepository {
89110
final List<AccountOnlineIp> records = new ArrayList<>();
111+
int attempts;
112+
int failuresBeforeSuccess;
90113

91114
@Override
92115
public void upsertOnlineStatus(String accountNo,
@@ -100,6 +123,11 @@ public void upsertOnlineStatus(String accountNo,
100123
LocalDateTime createTime,
101124
LocalDateTime updateTime,
102125
LocalDateTime offlineThresholdTime) {
126+
attempts++;
127+
if (failuresBeforeSuccess > 0) {
128+
failuresBeforeSuccess--;
129+
throw new PessimisticLockException("Deadlock found when trying to get lock; try restarting transaction");
130+
}
103131
AccountOnlineIp record = new AccountOnlineIp();
104132
record.setAccountNo(accountNo);
105133
record.setClientIp(clientIp);

0 commit comments

Comments
 (0)