refactor: 清理 PositionPollingService 代码
- 移除多余的空行 - 移除调试日志
This commit is contained in:
+16
-18
@@ -21,29 +21,29 @@ import java.util.concurrent.CopyOnWriteArrayList
|
|||||||
class PositionPollingService(
|
class PositionPollingService(
|
||||||
private val accountService: AccountService
|
private val accountService: AccountService
|
||||||
) {
|
) {
|
||||||
|
|
||||||
private val logger = LoggerFactory.getLogger(PositionPollingService::class.java)
|
private val logger = LoggerFactory.getLogger(PositionPollingService::class.java)
|
||||||
|
|
||||||
@Value("\${position.polling.interval:2000}")
|
@Value("\${position.polling.interval:2000}")
|
||||||
private var pollingInterval: Long = 2000 // 轮训间隔(毫秒),默认2秒
|
private var pollingInterval: Long = 2000 // 轮训间隔(毫秒),默认2秒
|
||||||
|
|
||||||
// 订阅者列表(支持多个订阅者)
|
// 订阅者列表(支持多个订阅者)
|
||||||
private val subscribers = CopyOnWriteArrayList<(PositionListResponse) -> Unit>()
|
private val subscribers = CopyOnWriteArrayList<(PositionListResponse) -> Unit>()
|
||||||
|
|
||||||
// 最新仓位数据(用于丢弃机制)
|
// 最新仓位数据(用于丢弃机制)
|
||||||
@Volatile
|
@Volatile
|
||||||
private var latestPositions: PositionListResponse? = null
|
private var latestPositions: PositionListResponse? = null
|
||||||
|
|
||||||
// 协程作用域和任务
|
// 协程作用域和任务
|
||||||
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||||
private var pollingJob: Job? = null
|
private var pollingJob: Job? = null
|
||||||
|
|
||||||
// 事件分发协程(使用专门的线程,避免阻塞轮训)
|
// 事件分发协程(使用专门的线程,避免阻塞轮训)
|
||||||
private val eventDispatcherScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
private val eventDispatcherScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
||||||
|
|
||||||
// 同步锁,确保轮询任务的启动和停止是线程安全的
|
// 同步锁,确保轮询任务的启动和停止是线程安全的
|
||||||
private val lock = Any()
|
private val lock = Any()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 初始化服务(后端启动时直接启动轮训)
|
* 初始化服务(后端启动时直接启动轮训)
|
||||||
*/
|
*/
|
||||||
@@ -52,7 +52,7 @@ class PositionPollingService(
|
|||||||
logger.info("PositionPollingService 初始化,启动仓位轮训任务,轮训间隔: ${pollingInterval}ms")
|
logger.info("PositionPollingService 初始化,启动仓位轮训任务,轮训间隔: ${pollingInterval}ms")
|
||||||
startPolling()
|
startPolling()
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 清理资源
|
* 清理资源
|
||||||
*/
|
*/
|
||||||
@@ -66,7 +66,7 @@ class PositionPollingService(
|
|||||||
scope.cancel()
|
scope.cancel()
|
||||||
eventDispatcherScope.cancel()
|
eventDispatcherScope.cancel()
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 订阅仓位事件
|
* 订阅仓位事件
|
||||||
* @param callback 回调函数,接收最新的仓位数据
|
* @param callback 回调函数,接收最新的仓位数据
|
||||||
@@ -78,7 +78,7 @@ class PositionPollingService(
|
|||||||
latestPositions?.let { callback(it) }
|
latestPositions?.let { callback(it) }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 取消订阅仓位事件
|
* 取消订阅仓位事件
|
||||||
*/
|
*/
|
||||||
@@ -87,7 +87,7 @@ class PositionPollingService(
|
|||||||
subscribers.remove(callback)
|
subscribers.remove(callback)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 启动轮训任务
|
* 启动轮训任务
|
||||||
*/
|
*/
|
||||||
@@ -95,7 +95,7 @@ class PositionPollingService(
|
|||||||
synchronized(lock) {
|
synchronized(lock) {
|
||||||
// 如果已经有轮训任务在运行,先取消
|
// 如果已经有轮训任务在运行,先取消
|
||||||
pollingJob?.cancel()
|
pollingJob?.cancel()
|
||||||
|
|
||||||
// 启动新的轮训任务
|
// 启动新的轮训任务
|
||||||
pollingJob = scope.launch {
|
pollingJob = scope.launch {
|
||||||
while (isActive) {
|
while (isActive) {
|
||||||
@@ -109,7 +109,7 @@ class PositionPollingService(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 轮训仓位数据并发布事件
|
* 轮训仓位数据并发布事件
|
||||||
* 使用专门的线程分发事件,避免阻塞轮训
|
* 使用专门的线程分发事件,避免阻塞轮训
|
||||||
@@ -123,7 +123,7 @@ class PositionPollingService(
|
|||||||
if (positions != null) {
|
if (positions != null) {
|
||||||
// 更新最新数据(丢弃旧数据,只保留最新的)
|
// 更新最新数据(丢弃旧数据,只保留最新的)
|
||||||
latestPositions = positions
|
latestPositions = positions
|
||||||
|
|
||||||
// 在专门的线程中分发事件,避免阻塞轮训
|
// 在专门的线程中分发事件,避免阻塞轮训
|
||||||
eventDispatcherScope.launch {
|
eventDispatcherScope.launch {
|
||||||
try {
|
try {
|
||||||
@@ -131,7 +131,7 @@ class PositionPollingService(
|
|||||||
val currentSubscribers = synchronized(lock) {
|
val currentSubscribers = synchronized(lock) {
|
||||||
subscribers.toList() // 复制列表,避免并发修改
|
subscribers.toList() // 复制列表,避免并发修改
|
||||||
}
|
}
|
||||||
|
|
||||||
currentSubscribers.forEach { callback ->
|
currentSubscribers.forEach { callback ->
|
||||||
try {
|
try {
|
||||||
callback(positions)
|
callback(positions)
|
||||||
@@ -139,8 +139,6 @@ class PositionPollingService(
|
|||||||
logger.error("通知订阅者失败: ${e.message}", e)
|
logger.error("通知订阅者失败: ${e.message}", e)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.debug("发布仓位数据事件: currentPositions=${positions.currentPositions.size}, historyPositions=${positions.historyPositions.size}, subscribers=${currentSubscribers.size}")
|
|
||||||
} catch (e: Exception) {
|
} catch (e: Exception) {
|
||||||
logger.error("分发仓位数据事件失败: ${e.message}", e)
|
logger.error("分发仓位数据事件失败: ${e.message}", e)
|
||||||
}
|
}
|
||||||
|
|||||||
+16
@@ -1,5 +1,7 @@
|
|||||||
package com.wrbug.polymarketbot.service.copytrading.monitor
|
package com.wrbug.polymarketbot.service.copytrading.monitor
|
||||||
|
|
||||||
|
import com.github.benmanes.caffeine.cache.Cache
|
||||||
|
import com.github.benmanes.caffeine.cache.Caffeine
|
||||||
import com.google.gson.JsonNull
|
import com.google.gson.JsonNull
|
||||||
import com.wrbug.polymarketbot.api.*
|
import com.wrbug.polymarketbot.api.*
|
||||||
import com.wrbug.polymarketbot.entity.Leader
|
import com.wrbug.polymarketbot.entity.Leader
|
||||||
@@ -12,6 +14,7 @@ import okhttp3.OkHttpClient
|
|||||||
import org.slf4j.LoggerFactory
|
import org.slf4j.LoggerFactory
|
||||||
import org.springframework.stereotype.Service
|
import org.springframework.stereotype.Service
|
||||||
import java.util.concurrent.ConcurrentHashMap
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
|
import java.util.concurrent.TimeUnit
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 链上 WebSocket 监听服务
|
* 链上 WebSocket 监听服务
|
||||||
@@ -30,6 +33,11 @@ class OnChainWsService(
|
|||||||
// 存储需要监听的Leader:leaderId -> Leader
|
// 存储需要监听的Leader:leaderId -> Leader
|
||||||
private val monitoredLeaders = ConcurrentHashMap<Long, Leader>()
|
private val monitoredLeaders = ConcurrentHashMap<Long, Leader>()
|
||||||
|
|
||||||
|
// 存储已处理的交易哈希,用于去重(LRU 缓存,保留最近 100 条)
|
||||||
|
private val processedTxHashes: Cache<String, Long> = Caffeine.newBuilder()
|
||||||
|
.maximumSize(100)
|
||||||
|
.build()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 启动链上 WebSocket 监听
|
* 启动链上 WebSocket 监听
|
||||||
* 通过统一服务订阅所有 Leader
|
* 通过统一服务订阅所有 Leader
|
||||||
@@ -95,6 +103,14 @@ class OnChainWsService(
|
|||||||
) {
|
) {
|
||||||
val leader = monitoredLeaders[leaderId] ?: return
|
val leader = monitoredLeaders[leaderId] ?: return
|
||||||
|
|
||||||
|
// 根据 txHash 去重(使用原子操作避免竞态条件)
|
||||||
|
val currentTime = System.currentTimeMillis()
|
||||||
|
val existingTimestamp = processedTxHashes.asMap().putIfAbsent(txHash, currentTime)
|
||||||
|
if (existingTimestamp != null) {
|
||||||
|
logger.debug("交易已处理过,跳过: leaderId=$leaderId, txHash=$txHash, firstProcessedAt=$existingTimestamp")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
logger.debug("开始处理 Leader 交易: leaderId=$leaderId, txHash=$txHash, leaderAddress=${leader.leaderAddress}")
|
logger.debug("开始处理 Leader 交易: leaderId=$leaderId, txHash=$txHash, leaderAddress=${leader.leaderAddress}")
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
|||||||
Reference in New Issue
Block a user