Golang实现灵活超时控制的协程用法

要实现灵活的协程超时控制，可以通过动态刷新超时计时器来实现。以下是两种实用的解决方案：

方案一：使用可重置的计时器

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func main() {
	fmt.Println("handling request")
	workers := 10

	var wg sync.WaitGroup
	wg.Add(workers)

	// 创建可取消的上下文
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 创建可重置的计时器
	resetTimer := time.NewTimer(5 * time.Second)
	defer resetTimer.Stop()

	// 监控工作协程完成情况
	done := make(chan int, workers)
	
	for i := 1; i <= workers; i++ {
		go worker(ctx, i, &wg, done)
	}

	// 协程完成监控器
	go func() {
		activeWorkers := workers
		for {
			select {
			case <-done:
				activeWorkers--
				// 重置计时器，只要有协程仍在活动
				if activeWorkers > 0 {
					if !resetTimer.Stop() {
						<-resetTimer.C
					}
					resetTimer.Reset(5 * time.Second)
				}
			case <-resetTimer.C:
				// 超时发生，取消所有协程
				cancel()
				return
			}
		}
	}()

	wg.Wait()
	fmt.Println("All workers terminated")
}

func worker(ctx context.Context, index int, wg *sync.WaitGroup, done chan<- int) {
	defer wg.Done()
	defer func() { done <- index }()

	select {
	case <-ctx.Done():
		fmt.Println("Worker", index, "timed out")
		return
	case <-time.After(time.Duration(index) * time.Second):
		fmt.Println("Worker", index, "finished")
	}
}

方案二：使用活动心跳机制

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func main() {
	fmt.Println("handling request")
	workers := 10

	var wg sync.WaitGroup
	wg.Add(workers)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 心跳通道
	heartbeat := make(chan struct{}, workers)
	
	// 超时监控器
	go func() {
		timeout := 5 * time.Second
		timer := time.NewTimer(timeout)
		defer timer.Stop()

		for {
			select {
			case <-heartbeat:
				// 收到心跳，重置计时器
				if !timer.Stop() {
					<-timer.C
				}
				timer.Reset(timeout)
			case <-timer.C:
				// 超时发生
				cancel()
				return
			}
		}
	}()

	for i := 1; i <= workers; i++ {
		go workerWithHeartbeat(ctx, i, &wg, heartbeat)
	}

	wg.Wait()
	fmt.Println("All workers terminated")
}

func workerWithHeartbeat(ctx context.Context, index int, wg *sync.WaitGroup, heartbeat chan<- struct{}) {
	defer wg.Done()

	// 定期发送心跳
	heartbeatTicker := time.NewTicker(1 * time.Second)
	defer heartbeatTicker.Stop()

	// 工作完成信号
	done := make(chan struct{})

	go func() {
		time.Sleep(time.Duration(index) * time.Second)
		close(done)
	}()

	for {
		select {
		case <-ctx.Done():
			fmt.Println("Worker", index, "timed out")
			return
		case <-heartbeatTicker.C:
			// 发送心跳
			select {
			case heartbeat <- struct{}{}:
			default:
			}
		case <-done:
			fmt.Println("Worker", index, "finished")
			return
		}
	}
}

方案三：使用进度报告机制

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type ProgressReporter interface {
	ReportProgress()
}

func main() {
	fmt.Println("handling request")
	workers := 10

	var wg sync.WaitGroup
	wg.Add(workers)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 进度监控器
	monitor := NewActivityMonitor(5*time.Second, cancel)
	defer monitor.Stop()

	for i := 1; i <= workers; i++ {
		go workerWithProgress(ctx, i, &wg, monitor)
	}

	wg.Wait()
	fmt.Println("All workers terminated")
}

type ActivityMonitor struct {
	timeout   time.Duration
	cancel    context.CancelFunc
	timer     *time.Timer
	progress  chan struct{}
	stop      chan struct{}
}

func NewActivityMonitor(timeout time.Duration, cancel context.CancelFunc) *ActivityMonitor {
	m := &ActivityMonitor{
		timeout:  timeout,
		cancel:   cancel,
		progress: make(chan struct{}, 100),
		stop:     make(chan struct{}),
		timer:    time.NewTimer(timeout),
	}

	go m.monitor()
	return m
}

func (m *ActivityMonitor) ReportProgress() {
	select {
	case m.progress <- struct{}{}:
	default:
	}
}

func (m *ActivityMonitor) monitor() {
	for {
		select {
		case <-m.progress:
			if !m.timer.Stop() {
				<-m.timer.C
			}
			m.timer.Reset(m.timeout)
		case <-m.timer.C:
			m.cancel()
			return
		case <-m.stop:
			return
		}
	}
}

func (m *ActivityMonitor) Stop() {
	close(m.stop)
	m.timer.Stop()
}

func workerWithProgress(ctx context.Context, index int, wg *sync.WaitGroup, monitor *ActivityMonitor) {
	defer wg.Done()

	// 模拟工作进度
	for i := 0; i < index; i++ {
		select {
		case <-ctx.Done():
			fmt.Println("Worker", index, "timed out")
			return
		default:
			// 报告进度
			monitor.ReportProgress()
			time.Sleep(1 * time.Second)
		}
	}
	fmt.Println("Worker", index, "finished")
}

这些方案都允许在协程活动时重置超时计时器，确保只有在所有协程都真正不活动时才触发超时。方案一通过协程完成事件重置计时器，方案二使用心跳机制，方案三提供更细粒度的进度报告。