diff --git a/common/progress_manager.go b/common/progress_manager.go index fc794f6..797e654 100644 --- a/common/progress_manager.go +++ b/common/progress_manager.go @@ -28,8 +28,8 @@ ProgressManager.go - 固定底部进度条管理器 type ProgressManager struct { mu sync.RWMutex enabled bool - total int64 - current int64 + total atomic.Int64 + current atomic.Int64 description string startTime time.Time isActive bool @@ -117,8 +117,8 @@ func (pm *ProgressManager) InitProgress(total int64, description string) { pm.mu.Lock() defer pm.mu.Unlock() - pm.total = total - pm.current = 0 + pm.total.Store(total) + pm.current.Store(0) pm.description = description pm.startTime = time.Now() pm.isActive = true @@ -144,9 +144,9 @@ func (pm *ProgressManager) UpdateProgress(increment int64) { } // 原子累加，避免高并发下的锁竞争 - newCurrent := atomic.AddInt64(&pm.current, increment) - if newCurrent > pm.total { - atomic.StoreInt64(&pm.current, pm.total) + newCurrent := pm.current.Add(increment) + if newCurrent > pm.total.Load() { + pm.current.Store(pm.total.Load()) } // 节流渲染：距上次渲染不足 50ms 则跳过 @@ -178,7 +178,7 @@ func (pm *ProgressManager) FinishProgress() { pm.mu.Lock() defer pm.mu.Unlock() - atomic.StoreInt64(&pm.current, pm.total) + pm.current.Store(pm.total.Load()) pm.renderProgress() // 停止活跃指示器 @@ -219,7 +219,7 @@ func (pm *ProgressManager) generateProgressBar() string { // 获取发包统计 packetInfo := pm.getPacketInfo() - if pm.total == 0 { + if pm.total.Load() == 0 { spinner := pm.getActivityIndicator() base := fmt.Sprintf("%s %s %s", pm.description, spinner, i18n.GetText("progress_waiting")) if packetInfo != "" { @@ -228,9 +228,9 @@ func (pm *ProgressManager) generateProgressBar() string { return base } - percentage := float64(atomic.LoadInt64(&pm.current)) / float64(pm.total) * 100 + percentage := float64(pm.current.Load()) / float64(pm.total.Load()) * 100 elapsed := time.Since(pm.startTime) - current := atomic.LoadInt64(&pm.current) + current := pm.current.Load() // 计算速度 speed := float64(current) / elapsed.Seconds() @@ -241,8 +241,8 @@ func (pm *ProgressManager) generateProgressBar() string { // 计算预估剩余时间 var eta string - if current > 0 && current < pm.total { - totalTime := elapsed * time.Duration(pm.total) / time.Duration(current) + if current > 0 && current < pm.total.Load() { + totalTime := elapsed * time.Duration(pm.total.Load()) / time.Duration(current) remaining := totalTime - elapsed if remaining > 0 { eta = fmt.Sprintf(" ETA:%s", formatDuration(remaining)) @@ -254,7 +254,7 @@ func (pm *ProgressManager) generateProgressBar() string { // 计算固定部分的宽度 fixedPart := fmt.Sprintf("%s %s %5.1f%% [] (%d/%d)%s%s %s", - pm.description, spinner, percentage, current, pm.total, speedStr, eta, packetInfo) + pm.description, spinner, percentage, current, pm.total.Load(), speedStr, eta, packetInfo) fixedWidth := displayWidth(fixedPart) // 计算进度条槽位可用宽度（预留2字符余量） @@ -281,7 +281,7 @@ func (pm *ProgressManager) generateProgressBar() string { // 构建最终进度条 result := fmt.Sprintf("%s %s %5.1f%% %s (%d/%d)%s%s", - pm.description, spinner, percentage, bar, current, pm.total, speedStr, eta) + pm.description, spinner, percentage, bar, current, pm.total.Load(), speedStr, eta) if packetInfo != "" { result += " " + packetInfo @@ -323,10 +323,10 @@ func (pm *ProgressManager) showCompletionInfo() { durationMsg := i18n.GetText("progress_duration") if pm.noColor { fmt.Printf("[%s] %s %d/%d (%s: %s)\n", - doneMsg, completionMsg, pm.total, pm.total, durationMsg, formatDuration(elapsed)) + doneMsg, completionMsg, pm.total.Load(), pm.total.Load(), durationMsg, formatDuration(elapsed)) } else { fmt.Printf("%s[%s] %s %d/%d%s %s(%s: %s)%s\n", - AnsiGreen, doneMsg, completionMsg, pm.total, pm.total, AnsiReset, + AnsiGreen, doneMsg, completionMsg, pm.total.Load(), pm.total.Load(), AnsiReset, AnsiGray, durationMsg, formatDuration(elapsed), AnsiReset) } } @@ -478,10 +478,10 @@ func (pm *ProgressManager) GetPercent() float64 { pm.mu.RLock() defer pm.mu.RUnlock() - if !pm.isActive || pm.total == 0 { + if !pm.isActive || pm.total.Load() == 0 { return 0 } - return float64(atomic.LoadInt64(&pm.current)) / float64(pm.total) * 100 + return float64(pm.current.Load()) / float64(pm.total.Load()) * 100 } // ============================================================================= @@ -522,8 +522,8 @@ func (pm *ProgressManager) renderProgressUnsafe() { // 计算当前百分比（避免除零） currentPercent := 0 - if pm.total > 0 { - currentPercent = int((atomic.LoadInt64(&pm.current) * 100) / pm.total) + if pm.total.Load() > 0 { + currentPercent = int((pm.current.Load() * 100) / pm.total.Load()) } // 只在百分比变化时更新，减少不必要的渲染 @@ -642,8 +642,8 @@ ConcurrencyMonitor - 并发监控器 // ConcurrencyMonitor 并发监控器 type ConcurrencyMonitor struct { // 主扫描器层级 - activePluginTasks int64 // 当前活跃的插件任务数 - totalPluginTasks int64 // 总插件任务数 + activePluginTasks atomic.Int64 // 当前活跃的插件任务数 + totalPluginTasks atomic.Int64 // 总插件任务数 // 插件内连接层级已移除 - 原代码为死代码，无任何调用者 } @@ -658,10 +658,7 @@ var ( // GetConcurrencyMonitor 获取全局并发监控器 func GetConcurrencyMonitor() *ConcurrencyMonitor { concurrencyMutex.Do(func() { - globalConcurrencyMonitor = &ConcurrencyMonitor{ - activePluginTasks: 0, - totalPluginTasks: 0, - } + globalConcurrencyMonitor = &ConcurrencyMonitor{} }) return globalConcurrencyMonitor } @@ -672,18 +669,18 @@ func GetConcurrencyMonitor() *ConcurrencyMonitor { // StartPluginTask 开始插件任务 func (m *ConcurrencyMonitor) StartPluginTask() { - atomic.AddInt64(&m.activePluginTasks, 1) - atomic.AddInt64(&m.totalPluginTasks, 1) + m.activePluginTasks.Add(1) + m.totalPluginTasks.Add(1) } // FinishPluginTask 完成插件任务 func (m *ConcurrencyMonitor) FinishPluginTask() { - atomic.AddInt64(&m.activePluginTasks, -1) + m.activePluginTasks.Add(-1) } // GetPluginTaskStats 获取插件任务统计 func (m *ConcurrencyMonitor) GetPluginTaskStats() (active int64, total int64) { - return atomic.LoadInt64(&m.activePluginTasks), atomic.LoadInt64(&m.totalPluginTasks) + return m.activePluginTasks.Load(), m.totalPluginTasks.Load() } // ============================================================================= diff --git a/common/proxy/httpdialer.go b/common/proxy/httpdialer.go index 430a2d6..6ac91ed 100644 --- a/common/proxy/httpdialer.go +++ b/common/proxy/httpdialer.go @@ -7,7 +7,6 @@ import ( "fmt" "net" "net/http" - "sync/atomic" "time" ) @@ -24,33 +23,27 @@ func (h *httpDialer) Dial(network, address string) (net.Conn, error) { func (h *httpDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) { start := time.Now() - atomic.AddInt64(&h.stats.TotalConnections, 1) + h.stats.addTotal(1) // 连接到HTTP代理服务器 proxyConn, err := h.baseDial.DialContext(ctx, NetworkTCP, h.config.Address) if err != nil { - atomic.AddInt64(&h.stats.FailedConnections, 1) - h.stats.mu.Lock() - h.stats.LastError = err.Error() - h.stats.mu.Unlock() + h.stats.addFailed(1) + h.stats.setLastError(err.Error()) return nil, NewProxyError(ErrTypeConnection, ErrMsgHTTPConnFailed, ErrCodeHTTPConnFailed, err) } // 发送CONNECT请求 if err := h.sendConnectRequest(proxyConn, address); err != nil { _ = proxyConn.Close() // 错误处理路径，Close错误可忽略 - atomic.AddInt64(&h.stats.FailedConnections, 1) - h.stats.mu.Lock() - h.stats.LastError = err.Error() - h.stats.mu.Unlock() + h.stats.addFailed(1) + h.stats.setLastError(err.Error()) return nil, err } duration := time.Since(start) - h.stats.mu.Lock() - h.stats.LastConnectTime = start - h.stats.mu.Unlock() - atomic.AddInt64(&h.stats.ActiveConnections, 1) + h.stats.setLastConnectTime(start) + h.stats.addActive(1) h.updateAverageConnectTime(duration) return &trackedConn{ diff --git a/common/proxy/manager.go b/common/proxy/manager.go index 2080f2b..720c228 100644 --- a/common/proxy/manager.go +++ b/common/proxy/manager.go @@ -5,7 +5,6 @@ import ( "fmt" "net" "sync" - "sync/atomic" "time" "golang.org/x/net/proxy" @@ -127,19 +126,7 @@ func (m *manager) Stats() *ProxyStats { m.mu.RLock() defer m.mu.RUnlock() - m.stats.mu.Lock() - defer m.stats.mu.Unlock() - - return &ProxyStats{ - TotalConnections: atomic.LoadInt64(&m.stats.TotalConnections), - ActiveConnections: atomic.LoadInt64(&m.stats.ActiveConnections), - FailedConnections: atomic.LoadInt64(&m.stats.FailedConnections), - AverageConnectTime: m.stats.AverageConnectTime, - LastConnectTime: m.stats.LastConnectTime, - LastError: m.stats.LastError, - ProxyType: m.stats.ProxyType, - ProxyAddress: m.stats.ProxyAddress, - } + return m.stats.snapshot() } // createDirectDialer 创建直连拨号器 @@ -246,7 +233,7 @@ func (d *directDialer) Dial(network, address string) (net.Conn, error) { func (d *directDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) { start := time.Now() - atomic.AddInt64(&d.stats.TotalConnections, 1) + d.stats.addTotal(1) dialer := &net.Dialer{ Timeout: d.timeout, @@ -263,19 +250,15 @@ func (d *directDialer) DialContext(ctx context.Context, network, address string) duration := time.Since(start) - d.stats.mu.Lock() - d.stats.LastConnectTime = start - d.stats.mu.Unlock() + d.stats.setLastConnectTime(start) if err != nil { - atomic.AddInt64(&d.stats.FailedConnections, 1) - d.stats.mu.Lock() - d.stats.LastError = err.Error() - d.stats.mu.Unlock() + d.stats.addFailed(1) + d.stats.setLastError(err.Error()) return nil, NewProxyError(ErrTypeConnection, ErrMsgDirectConnFailed, ErrCodeDirectConnFailed, err) } - atomic.AddInt64(&d.stats.ActiveConnections, 1) + d.stats.addActive(1) d.updateAverageConnectTime(duration) return &trackedConn{ @@ -297,7 +280,7 @@ func (s *socks5Dialer) Dial(network, address string) (net.Conn, error) { func (s *socks5Dialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) { start := time.Now() - atomic.AddInt64(&s.stats.TotalConnections, 1) + s.stats.addTotal(1) // 创建一个带超时的上下文 dialCtx, cancel := context.WithTimeout(ctx, s.config.Timeout) @@ -325,27 +308,21 @@ func (s *socks5Dialer) DialContext(ctx context.Context, network, address string) select { case <-dialCtx.Done(): - atomic.AddInt64(&s.stats.FailedConnections, 1) - s.stats.mu.Lock() - s.stats.LastError = dialCtx.Err().Error() - s.stats.mu.Unlock() + s.stats.addFailed(1) + s.stats.setLastError(dialCtx.Err().Error()) return nil, NewProxyError(ErrTypeTimeout, ErrMsgSOCKS5ConnTimeout, ErrCodeSOCKS5ConnTimeout, dialCtx.Err()) case result := <-connChan: duration := time.Since(start) - s.stats.mu.Lock() - s.stats.LastConnectTime = start - s.stats.mu.Unlock() + s.stats.setLastConnectTime(start) if result.err != nil { - atomic.AddInt64(&s.stats.FailedConnections, 1) - s.stats.mu.Lock() - s.stats.LastError = result.err.Error() - s.stats.mu.Unlock() + s.stats.addFailed(1) + s.stats.setLastError(result.err.Error()) return nil, NewProxyError(ErrTypeConnection, ErrMsgSOCKS5ConnFailed, ErrCodeSOCKS5ConnFailed, result.err) } - atomic.AddInt64(&s.stats.ActiveConnections, 1) + s.stats.addActive(1) s.updateAverageConnectTime(duration) return &trackedConn{ diff --git a/common/proxy/tlsdialer.go b/common/proxy/tlsdialer.go index 3af4292..d4921f1 100644 --- a/common/proxy/tlsdialer.go +++ b/common/proxy/tlsdialer.go @@ -49,10 +49,8 @@ func (t *tlsDialerWrapper) DialTLSContext(ctx context.Context, network, address // 进行TLS握手 if err := tlsConn.Handshake(); err != nil { _ = tcpConn.Close() // TLS握手失败，Close错误可忽略 - atomic.AddInt64(&t.stats.FailedConnections, 1) - t.stats.mu.Lock() - t.stats.LastError = err.Error() - t.stats.mu.Unlock() + t.stats.addFailed(1) + t.stats.setLastError(err.Error()) return nil, NewProxyError(ErrTypeConnection, ErrMsgTLSHandshakeFailed, ErrCodeTLSHandshakeFailed, err) } @@ -84,16 +82,16 @@ func (t *tlsDialerWrapper) updateAverageConnectTime(duration time.Duration) { // trackedConn 带统计的连接 type trackedConn struct { + bytesSent atomic.Int64 + bytesRecv atomic.Int64 net.Conn - stats *ProxyStats - bytesSent int64 - bytesRecv int64 + stats *ProxyStats } func (tc *trackedConn) Read(b []byte) (n int, err error) { n, err = tc.Conn.Read(b) if n > 0 { - atomic.AddInt64(&tc.bytesRecv, int64(n)) + tc.bytesRecv.Add(int64(n)) } return n, err } @@ -101,13 +99,13 @@ func (tc *trackedConn) Read(b []byte) (n int, err error) { func (tc *trackedConn) Write(b []byte) (n int, err error) { n, err = tc.Conn.Write(b) if n > 0 { - atomic.AddInt64(&tc.bytesSent, int64(n)) + tc.bytesSent.Add(int64(n)) } return n, err } func (tc *trackedConn) Close() error { - atomic.AddInt64(&tc.stats.ActiveConnections, -1) + tc.stats.addActive(-1) return tc.Conn.Close() } diff --git a/common/proxy/types.go b/common/proxy/types.go index adb8c50..985a5c2 100644 --- a/common/proxy/types.go +++ b/common/proxy/types.go @@ -96,9 +96,9 @@ type ProxyManager interface { // //nolint:revive // 保持与现有代码的向后兼容性 type ProxyStats struct { - TotalConnections int64 `json:"total_connections"` - ActiveConnections int64 `json:"active_connections"` - FailedConnections int64 `json:"failed_connections"` + TotalConnections int64 `json:"total_connections"` + ActiveConnections int64 `json:"active_connections"` + FailedConnections int64 `json:"failed_connections"` mu sync.Mutex `json:"-"` AverageConnectTime time.Duration `json:"average_connect_time"` LastConnectTime time.Time `json:"last_connect_time"` @@ -107,6 +107,52 @@ type ProxyStats struct { ProxyAddress string `json:"proxy_address"` } +func (s *ProxyStats) addTotal(delta int64) { + s.mu.Lock() + s.TotalConnections += delta + s.mu.Unlock() +} + +func (s *ProxyStats) addActive(delta int64) { + s.mu.Lock() + s.ActiveConnections += delta + s.mu.Unlock() +} + +func (s *ProxyStats) addFailed(delta int64) { + s.mu.Lock() + s.FailedConnections += delta + s.mu.Unlock() +} + +func (s *ProxyStats) setLastConnectTime(t time.Time) { + s.mu.Lock() + s.LastConnectTime = t + s.mu.Unlock() +} + +func (s *ProxyStats) setLastError(err string) { + s.mu.Lock() + s.LastError = err + s.mu.Unlock() +} + +func (s *ProxyStats) snapshot() *ProxyStats { + s.mu.Lock() + defer s.mu.Unlock() + + return &ProxyStats{ + TotalConnections: s.TotalConnections, + ActiveConnections: s.ActiveConnections, + FailedConnections: s.FailedConnections, + AverageConnectTime: s.AverageConnectTime, + LastConnectTime: s.LastConnectTime, + LastError: s.LastError, + ProxyType: s.ProxyType, + ProxyAddress: s.ProxyAddress, + } +} + // ProxyError 代理错误类型 // //nolint:revive // 保持与现有代码的向后兼容性 diff --git a/common/state.go b/common/state.go index e8fcd00..da4dea5 100644 --- a/common/state.go +++ b/common/state.go @@ -23,17 +23,17 @@ state.go - 运行时状态管理 // State 扫描器运行时状态 - 线程安全 type State struct { // 计数器 - 原子操作 - packetCount int64 - tcpPacketCount int64 - tcpSuccessPacketCount int64 - tcpFailedPacketCount int64 - udpPacketCount int64 - httpPacketCount int64 - resourceExhaustedCount int64 + packetCount atomic.Int64 + tcpPacketCount atomic.Int64 + tcpSuccessPacketCount atomic.Int64 + tcpFailedPacketCount atomic.Int64 + udpPacketCount atomic.Int64 + httpPacketCount atomic.Int64 + resourceExhaustedCount atomic.Int64 // 任务计数 - end int64 - num int64 + end atomic.Int64 + num atomic.Int64 // 时间 startTime time.Time @@ -71,38 +71,38 @@ func NewState() *State { // IncrementPacketCount 增加总包计数 func (s *State) IncrementPacketCount() int64 { - return atomic.AddInt64(&s.packetCount, 1) + return s.packetCount.Add(1) } // IncrementTCPSuccessPacketCount 增加TCP成功连接包计数 func (s *State) IncrementTCPSuccessPacketCount() int64 { - atomic.AddInt64(&s.tcpSuccessPacketCount, 1) - atomic.AddInt64(&s.tcpPacketCount, 1) - return atomic.AddInt64(&s.packetCount, 1) + s.tcpSuccessPacketCount.Add(1) + s.tcpPacketCount.Add(1) + return s.packetCount.Add(1) } // IncrementTCPFailedPacketCount 增加TCP失败连接包计数 func (s *State) IncrementTCPFailedPacketCount() int64 { - atomic.AddInt64(&s.tcpFailedPacketCount, 1) - atomic.AddInt64(&s.tcpPacketCount, 1) - return atomic.AddInt64(&s.packetCount, 1) + s.tcpFailedPacketCount.Add(1) + s.tcpPacketCount.Add(1) + return s.packetCount.Add(1) } // IncrementUDPPacketCount 增加UDP包计数 func (s *State) IncrementUDPPacketCount() int64 { - atomic.AddInt64(&s.udpPacketCount, 1) - return atomic.AddInt64(&s.packetCount, 1) + s.udpPacketCount.Add(1) + return s.packetCount.Add(1) } // IncrementHTTPPacketCount 增加HTTP包计数 func (s *State) IncrementHTTPPacketCount() int64 { - atomic.AddInt64(&s.httpPacketCount, 1) - return atomic.AddInt64(&s.packetCount, 1) + s.httpPacketCount.Add(1) + return s.packetCount.Add(1) } // IncrementResourceExhaustedCount 增加资源耗尽错误计数 func (s *State) IncrementResourceExhaustedCount() { - atomic.AddInt64(&s.resourceExhaustedCount, 1) + s.resourceExhaustedCount.Add(1) } // ============================================================================= @@ -111,48 +111,48 @@ func (s *State) IncrementResourceExhaustedCount() { // GetPacketCount 获取总包计数 func (s *State) GetPacketCount() int64 { - return atomic.LoadInt64(&s.packetCount) + return s.packetCount.Load() } // GetTCPPacketCount 获取TCP包计数 func (s *State) GetTCPPacketCount() int64 { - return atomic.LoadInt64(&s.tcpPacketCount) + return s.tcpPacketCount.Load() } // GetTCPSuccessPacketCount 获取TCP成功连接包计数 func (s *State) GetTCPSuccessPacketCount() int64 { - return atomic.LoadInt64(&s.tcpSuccessPacketCount) + return s.tcpSuccessPacketCount.Load() } // GetTCPFailedPacketCount 获取TCP失败连接包计数 func (s *State) GetTCPFailedPacketCount() int64 { - return atomic.LoadInt64(&s.tcpFailedPacketCount) + return s.tcpFailedPacketCount.Load() } // GetUDPPacketCount 获取UDP包计数 func (s *State) GetUDPPacketCount() int64 { - return atomic.LoadInt64(&s.udpPacketCount) + return s.udpPacketCount.Load() } // GetHTTPPacketCount 获取HTTP包计数 func (s *State) GetHTTPPacketCount() int64 { - return atomic.LoadInt64(&s.httpPacketCount) + return s.httpPacketCount.Load() } // GetResourceExhaustedCount 获取资源耗尽错误计数 func (s *State) GetResourceExhaustedCount() int64 { - return atomic.LoadInt64(&s.resourceExhaustedCount) + return s.resourceExhaustedCount.Load() } // ResetPacketCounters 重置所有包计数器 func (s *State) ResetPacketCounters() { - atomic.StoreInt64(&s.packetCount, 0) - atomic.StoreInt64(&s.tcpPacketCount, 0) - atomic.StoreInt64(&s.tcpSuccessPacketCount, 0) - atomic.StoreInt64(&s.tcpFailedPacketCount, 0) - atomic.StoreInt64(&s.udpPacketCount, 0) - atomic.StoreInt64(&s.httpPacketCount, 0) - atomic.StoreInt64(&s.resourceExhaustedCount, 0) + s.packetCount.Store(0) + s.tcpPacketCount.Store(0) + s.tcpSuccessPacketCount.Store(0) + s.tcpFailedPacketCount.Store(0) + s.udpPacketCount.Store(0) + s.httpPacketCount.Store(0) + s.resourceExhaustedCount.Store(0) } // ============================================================================= @@ -161,32 +161,32 @@ func (s *State) ResetPacketCounters() { // GetEnd 获取结束计数 func (s *State) GetEnd() int64 { - return atomic.LoadInt64(&s.end) + return s.end.Load() } // GetNum 获取数量计数 func (s *State) GetNum() int64 { - return atomic.LoadInt64(&s.num) + return s.num.Load() } // IncrementEnd 增加结束计数 func (s *State) IncrementEnd() int64 { - return atomic.AddInt64(&s.end, 1) + return s.end.Add(1) } // IncrementNum 增加数量计数 func (s *State) IncrementNum() int64 { - return atomic.AddInt64(&s.num, 1) + return s.num.Add(1) } // SetEnd 设置结束计数 func (s *State) SetEnd(val int64) { - atomic.StoreInt64(&s.end, val) + s.end.Store(val) } // SetNum 设置数量计数 func (s *State) SetNum(val int64) { - atomic.StoreInt64(&s.num, val) + s.num.Store(val) } // ============================================================================= @@ -271,10 +271,10 @@ type PerfStatsData struct { func (s *State) GetPerfStats() PerfStatsData { duration := time.Since(s.startTime) durationMs := duration.Milliseconds() - totalPackets := atomic.LoadInt64(&s.packetCount) - tcpSuccess := atomic.LoadInt64(&s.tcpSuccessPacketCount) - tcpFailed := atomic.LoadInt64(&s.tcpFailedPacketCount) - tcpTotal := atomic.LoadInt64(&s.tcpPacketCount) + totalPackets := s.packetCount.Load() + tcpSuccess := s.tcpSuccessPacketCount.Load() + tcpFailed := s.tcpFailedPacketCount.Load() + tcpTotal := s.tcpPacketCount.Load() var pps float64 if durationMs > 0 { @@ -291,13 +291,13 @@ func (s *State) GetPerfStats() PerfStatsData { TCPPackets: tcpTotal, TCPSuccess: tcpSuccess, TCPFailed: tcpFailed, - UDPPackets: atomic.LoadInt64(&s.udpPacketCount), - HTTPPackets: atomic.LoadInt64(&s.httpPacketCount), - ResourceExhausted: atomic.LoadInt64(&s.resourceExhaustedCount), + UDPPackets: s.udpPacketCount.Load(), + HTTPPackets: s.httpPacketCount.Load(), + ResourceExhausted: s.resourceExhaustedCount.Load(), ScanDurationMs: durationMs, PacketsPerSecond: pps, SuccessRate: successRate, - TargetsScanned: atomic.LoadInt64(&s.num), + TargetsScanned: s.num.Load(), } } diff --git a/core/adaptive_pool.go b/core/adaptive_pool.go index f700c86..d07041e 100644 --- a/core/adaptive_pool.go +++ b/core/adaptive_pool.go @@ -24,7 +24,7 @@ type AdaptivePool struct { // 监控参数 checkInterval time.Duration - lastCheckNano int64 // 原子, UnixNano + lastCheckNano atomic.Int64 // UnixNano lastExhaustedCount int64 lastPacketCount int64 @@ -70,12 +70,12 @@ func (ap *AdaptivePool) Invoke(task interface{}) error { // maybeAdjust 检查并可能调整线程池大小 // 使用原子 CAS 进行时间检查，99%+ 的调用零锁开销 func (ap *AdaptivePool) maybeAdjust() { - lastCheck := atomic.LoadInt64(&ap.lastCheckNano) + lastCheck := ap.lastCheckNano.Load() now := time.Now().UnixNano() if now-lastCheck < int64(ap.checkInterval) { return } - if !atomic.CompareAndSwapInt64(&ap.lastCheckNano, lastCheck, now) { + if !ap.lastCheckNano.CompareAndSwap(lastCheck, now) { return // 其他 goroutine 已在检查 } diff --git a/core/port_scan.go b/core/port_scan.go index 7ac944e..64cd037 100644 --- a/core/port_scan.go +++ b/core/port_scan.go @@ -210,7 +210,7 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout // 初始化并发控制 to := time.Duration(timeout) * time.Second adaptiveTO := NewAdaptiveTimeout(to) - var count int64 + var count atomic.Int64 collector := newResultCollector(stream) failedCollector := &failedPortCollector{} var wg sync.WaitGroup @@ -258,7 +258,7 @@ func EnhancedPortScan(ctx context.Context, hosts []string, ports string, timeout common.FinishProgressBar() } - session.LogInfo(i18n.Tr("port_scan_complete", count)) + session.LogInfo(i18n.Tr("port_scan_complete", count.Load())) // 检查扫描失败率，如果过高则警告用户 resourceErrors := state.GetResourceExhaustedCount() @@ -453,7 +453,7 @@ func buildServiceLogMessage(addr string, serviceInfo *ServiceInfo, isWeb bool) s } // scanSinglePort 扫描单个端口并进行服务识别（重构后的简洁版本） -func scanSinglePort(ctx context.Context, host string, port int, addr string, adaptiveTO *AdaptiveTimeout, count *int64, collector *resultCollector, failedCollector *failedPortCollector, session *common.ScanSession) { +func scanSinglePort(ctx context.Context, host string, port int, addr string, adaptiveTO *AdaptiveTimeout, count *atomic.Int64, collector *resultCollector, failedCollector *failedPortCollector, session *common.ScanSession) { config := session.Config timeout := adaptiveTO.Timeout() // 步骤1：建立连接 @@ -486,7 +486,7 @@ func scanSinglePort(ctx context.Context, host string, port int, addr string, ada } // 步骤2：记录开放端口 - atomic.AddInt64(count, 1) + count.Add(1) collector.Add(addr) saveOpenPort(session, host, port)