Go语言中实现并发定时轮询与动态列表管理

func (obj *MyObj) Poll() {
    for ;; {
        for _, url := range obj.UrlList {
            // 下载URL内容并处理
            // harvest(url)
        }
        time.Sleep(30 * time.Minute)
    }
}

// 在其他函数中启动
// go obj.Poll()

type harvester struct {
    ticker *time.Ticker // 用于周期性触发轮询
    add    chan string  // 用于接收新的URL添加请求
    urls   []string     // 当前需要轮询的URL列表
}

func newHarvester() *harvester {
    rv := &harvester{
        ticker: time.NewTicker(time.Minute * 30), // 设置30分钟的轮询间隔
        add:    make(chan string),                 // 创建URL添加通道
        urls:   []string{},                        // 初始化为空URL列表
    }
    go rv.run() // 启动核心的运行goroutine
    return rv
}

func (h *harvester) run() {
    for {
        select {
        case <-h.ticker.C:
            // 当定时器触发时，执行URL轮询
            for _, u := range h.urls {
                // 模拟URL采集操作
                harvest(u)
            }
        case u := <-h.add:
            // 当有新的URL通过通道发送过来时，添加到URL列表中
            h.urls = append(h.urls, u)
        }
    }
}

// 模拟URL采集函数
func harvest(url string) {
    // 实际的下载和处理逻辑
    fmt.Printf("Harvesting URL: %s at %s\n", url, time.Now().Format("15:04:05"))
}

func (h *harvester) AddURL(u string) {
    // 将新的URL发送到add通道，由run goroutine安全处理
    h.add <- u
}

package main

import (
    "fmt"
    "time"
)

// harvester 结构体定义
type harvester struct {
    ticker *time.Ticker // 周期性触发器
    add    chan string  // 新URL添加通道
    urls   []string     // 当前URL列表
}

// newHarvester 创建并初始化一个新的harvester实例
func newHarvester() *harvester {
    rv := &harvester{
        ticker: time.NewTicker(time.Minute * 1), // 示例中设置为1分钟，方便测试
        add:    make(chan string),
        urls:   []string{},
    }
    go rv.run() // 启动核心运行goroutine
    return rv
}

// run 方法包含harvester的核心并发逻辑
func (h *harvester) run() {
    for {
        select {
        case <-h.ticker.C:
            // 当定时器触发时，执行URL轮询
            fmt.Println("\n--- Initiating URL harvest ---")
            if len(h.urls) == 0 {
                fmt.Println("No URLs to harvest yet.")
            }
            for _, u := range h.urls {
                harvest(u)
            }
            fmt.Println("--- URL harvest complete ---\n")
        case u := <-h.add:
            // 当有新的URL通过通道发送过来时，添加到URL列表中
            h.urls = append(h.urls, u)
            fmt.Printf("Added new URL: %s. Current URLs: %v\n", u, h.urls)
        }
    }
}

// AddURL 方法用于安全地向harvester添加新的URL
func (h *harvester) AddURL(u string) {
    h.add <- u
}

// harvest 模拟URL采集操作
func harvest(url string) {
    // 实际的下载和处理逻辑，这里仅作打印
    fmt.Printf("  Harvesting URL: %s at %s\n", url, time.Now().Format("15:04:05"))
    time.Sleep(time.Millisecond * 100) // 模拟网络延迟
}

func main() {
    // 创建一个harvester实例
    h := newHarvester()

    // 初始添加一些URL
    h.AddURL("http://example.com/page1")
    h.AddURL("http://example.com/page2")

    // 模拟程序运行一段时间
    fmt.Println("Harvester started. Initial URLs added.")
    fmt.Println("Waiting for first harvest cycle (1 minute)...")

    // 在一段时间后动态添加更多URL
    time.Sleep(time.Second * 30) // 等待30秒
    h.AddURL("http://example.com/new-page3")
    h.AddURL("http://example.com/new-page4")

    // 继续等待，观察后续的轮询是否包含新URL
    time.Sleep(time.Minute * 2) // 等待2分钟，至少经过两次轮询周期

    // 模拟程序关闭，停止ticker (重要：在实际应用中，需要更完善的优雅关闭机制)
    h.ticker.Stop()
    fmt.Println("Harvester stopped.")
}