proxy_expiry_service.go 1.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
  1. package service
  2. import (
  3. "context"
  4. "log"
  5. "sync"
  6. "time"
  7. )
  8. // ProxyExpiryService 周期扫描到期代理并把绑定账号改投备用/直连。
  9. type ProxyExpiryService struct {
  10. proxyRepo ProxyRepository
  11. interval time.Duration
  12. stopCh chan struct{}
  13. stopOnce sync.Once
  14. wg sync.WaitGroup
  15. }
  16. func NewProxyExpiryService(proxyRepo ProxyRepository, interval time.Duration) *ProxyExpiryService {
  17. return &ProxyExpiryService{proxyRepo: proxyRepo, interval: interval, stopCh: make(chan struct{})}
  18. }
  19. func (s *ProxyExpiryService) Start() {
  20. if s == nil || s.proxyRepo == nil || s.interval <= 0 {
  21. return
  22. }
  23. s.wg.Add(1)
  24. go func() {
  25. defer s.wg.Done()
  26. ticker := time.NewTicker(s.interval)
  27. defer ticker.Stop()
  28. s.runOnce()
  29. for {
  30. select {
  31. case <-ticker.C:
  32. s.runOnce()
  33. case <-s.stopCh:
  34. return
  35. }
  36. }
  37. }()
  38. }
  39. func (s *ProxyExpiryService) Stop() {
  40. if s == nil {
  41. return
  42. }
  43. s.stopOnce.Do(func() { close(s.stopCh) })
  44. s.wg.Wait()
  45. }
  46. func (s *ProxyExpiryService) runOnce() {
  47. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  48. defer cancel()
  49. changed, err := s.proxyRepo.SweepExpiredProxies(ctx, time.Now())
  50. if err != nil {
  51. log.Printf("[ProxyExpiry] sweep expired proxies failed: %v", err)
  52. return
  53. }
  54. if changed > 0 {
  55. log.Printf("[ProxyExpiry] re-routed %d accounts off expired proxies", changed)
  56. }
  57. }