Golang中如何实现net/http与并发调度器的同时运行

在Golang中实现高并发HTTP请求时，端口耗尽是常见问题。根本原因在于每个HTTP客户端连接都会占用一个本地端口，当并发量极大时，会快速耗尽可用端口。以下是针对你问题的解决方案：

1. 使用连接池和HTTP客户端复用

创建全局HTTP客户端并复用连接，而不是为每个请求创建新客户端：

var (
    httpClient *http.Client
    once sync.Once
)

func getHTTPClient() *http.Client {
    once.Do(func() {
        httpClient = &http.Client{
            Transport: &http.Transport{
                MaxIdleConns:        1000,
                MaxIdleConnsPerHost: 1000,
                MaxConnsPerHost:     1000,
                IdleConnTimeout:     90 * time.Second,
                TLSHandshakeTimeout: 10 * time.Second,
                DialContext: (&net.Dialer{
                    Timeout:   30 * time.Second,
                    KeepAlive: 30 * time.Second,
                }).DialContext,
            },
            Timeout: 30 * time.Second,
        }
    })
    return httpClient
}

2. 使用工作池控制并发

创建固定大小的goroutine池来控制并发数量：

type Job struct {
    MerchantID int
    URL        string
}

func worker(id int, jobs <-chan Job, results chan<- error) {
    client := getHTTPClient()
    for job := range jobs {
        err := processMerchant(client, job)
        results <- err
    }
}

func processMerchant(client *http.Client, job Job) error {
    req, err := http.NewRequest("GET", job.URL, nil)
    if err != nil {
        return err
    }
    
    resp, err := client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    
    _, err = io.Copy(io.Discard, resp.Body)
    return err
}

func ScheduleMerchantTasks(merchants []Merchant, workerCount int) {
    jobs := make(chan Job, len(merchants))
    results := make(chan error, len(merchants))
    
    // 启动worker
    for w := 1; w <= workerCount; w++ {
        go worker(w, jobs, results)
    }
    
    // 发送任务
    for _, m := range merchants {
        jobs <- Job{
            MerchantID: m.ID,
            URL:        fmt.Sprintf("http://localhost:8000/public/print-test?num=%d", m.ID),
        }
    }
    close(jobs)
    
    // 收集结果
    for i := 0; i < len(merchants); i++ {
        err := <-results
        if err != nil {
            log.Printf("Error processing merchant: %v", err)
        }
    }
}

3. 使用semaphore控制并发goroutine数量

import "golang.org/x/sync/semaphore"

func ProcessMerchantsConcurrently(merchants []Merchant, maxConcurrent int) {
    sem := semaphore.NewWeighted(int64(maxConcurrent))
    var wg sync.WaitGroup
    ctx := context.Background()
    
    for _, m := range merchants {
        wg.Add(1)
        go func(merchant Merchant) {
            defer wg.Done()
            
            // 获取信号量，控制并发
            if err := sem.Acquire(ctx, 1); err != nil {
                log.Printf("Failed to acquire semaphore: %v", err)
                return
            }
            defer sem.Release(1)
            
            // 处理商户任务
            processMerchantTask(merchant)
        }(m)
    }
    
    wg.Wait()
}

func processMerchantTask(m Merchant) {
    client := getHTTPClient()
    
    // 调用调度器1
    err := callScheduler(client, fmt.Sprintf("http://localhost:3333/scheduler1/%d", m.ID))
    if err != nil {
        log.Printf("Scheduler1 error for merchant %d: %v", m.ID, err)
    }
    
    // 调用调度器2
    err = callScheduler(client, fmt.Sprintf("http://localhost:3333/scheduler2/%d", m.ID))
    if err != nil {
        log.Printf("Scheduler2 error for merchant %d: %v", m.ID, err)
    }
}

4. 定时任务调度器实现

type MerchantScheduler struct {
    client      *http.Client
    merchantMap map[int]*time.Ticker
    mu          sync.RWMutex
}

func NewMerchantScheduler() *MerchantScheduler {
    return &MerchantScheduler{
        client:      getHTTPClient(),
        merchantMap: make(map[int]*time.Ticker),
    }
}

func (ms *MerchantScheduler) ScheduleMerchant(merchantID int, interval time.Duration) {
    ms.mu.Lock()
    defer ms.mu.Unlock()
    
    // 如果已经存在定时器，先停止
    if ticker, exists := ms.merchantMap[merchantID]; exists {
        ticker.Stop()
    }
    
    ticker := time.NewTicker(interval)
    ms.merchantMap[merchantID] = ticker
    
    go func(id int, t *time.Ticker) {
        for range t.C {
            ms.executeMerchantTask(id)
        }
    }(merchantID, ticker)
}

func (ms *MerchantScheduler) executeMerchantTask(merchantID int) {
    // 使用工作池执行任务，避免创建过多goroutine
    go func() {
        urls := []string{
            fmt.Sprintf("http://localhost:3333/scheduler1/%d", merchantID),
            fmt.Sprintf("http://localhost:3333/scheduler2/%d", merchantID),
        }
        
        var wg sync.WaitGroup
        for _, url := range urls {
            wg.Add(1)
            go func(u string) {
                defer wg.Done()
                req, _ := http.NewRequest("GET", u, nil)
                resp, err := ms.client.Do(req)
                if err != nil {
                    log.Printf("Error calling %s: %v", u, err)
                    return
                }
                defer resp.Body.Close()
                io.Copy(io.Discard, resp.Body)
            }(url)
        }
        wg.Wait()
    }()
}

5. 完整的优化版本

package main

import (
    "context"
    "fmt"
    "io"
    "log"
    "net/http"
    "sync"
    "time"
    
    "golang.org/x/sync/semaphore"
)

const (
    maxConcurrentRequests = 500  // 根据系统调整
    requestTimeout        = 30 * time.Second
)

type Scheduler struct {
    client *http.Client
    sem    *semaphore.Weighted
}

func NewScheduler() *Scheduler {
    transport := &http.Transport{
        MaxIdleConns:        maxConcurrentRequests,
        MaxIdleConnsPerHost: maxConcurrentRequests,
        MaxConnsPerHost:     maxConcurrentRequests,
        IdleConnTimeout:     90 * time.Second,
    }
    
    return &Scheduler{
        client: &http.Client{
            Transport: transport,
            Timeout:   requestTimeout,
        },
        sem: semaphore.NewWeighted(maxConcurrentRequests),
    }
}

func (s *Scheduler) ProcessMerchant(merchantID int) error {
    ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
    defer cancel()
    
    // 获取信号量控制并发
    if err := s.sem.Acquire(ctx, 1); err != nil {
        return fmt.Errorf("failed to acquire semaphore: %w", err)
    }
    defer s.sem.Release(1)
    
    // 并发调用两个调度器
    var wg sync.WaitGroup
    errors := make(chan error, 2)
    
    urls := []string{
        fmt.Sprintf("http://localhost:3333/scheduler1/%d", merchantID),
        fmt.Sprintf("http://localhost:3333/scheduler2/%d", merchantID),
    }
    
    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            if err := s.makeRequest(u); err != nil {
                errors <- err
            }
        }(url)
    }
    
    wg.Wait()
    close(errors)
    
    // 返回第一个错误
    for err := range errors {
        return err
    }
    
    return nil
}

func (s *Scheduler) makeRequest(url string) error {
    req, err := http.NewRequest("GET", url, nil)
    if err != nil {
        return err
    }
    
    resp, err := s.client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    
    // 读取并丢弃响应体
    _, err = io.Copy(io.Discard, resp.Body)
    return err
}

// 定时执行任务
func (s *Scheduler) StartPeriodicTasks(merchants []int, interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()
    
    for range ticker.C {
        s.processBatch(merchants)
    }
}

func (s *Scheduler) processBatch(merchants []int) {
    var wg sync.WaitGroup
    for _, merchantID := range merchants {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            if err := s.ProcessMerchant(id); err != nil {
                log.Printf("Error processing merchant %d: %v", id, err)
            }
        }(merchantID)
    }
    wg.Wait()
}

关键优化点：

1. 复用HTTP客户端和连接池
2. 使用信号量控制最大并发数
3. 避免为每个请求创建新客户端
4. 合理设置连接超时和空闲超时
5. 使用context控制请求超时

这个架构可以处理成千上万的商户定时任务，而不会耗尽系统资源。通过控制并发数量和复用连接，可以避免端口耗尽和文件描述符不足的问题。