Go 语言并发采集淘宝商品数据：利用 API 实现高性能抓取

go get github.com/imroc/req/v3
go get github.com/tidwall/gjson
go get github.com/joho/godotenv
go get golang.org/x/sync/errgroup

package main

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"sync"
	"time"

	"github.com/joho/godotenv"
	"github.com/tidwall/gjson"
	"golang.org/x/sync/errgroup"
)

// 商品信息结构体
type Product struct {
	ID          string  `json:"id"`
	Title       string  `json:"title"`
	Price       string  `json:"price"`
	Sales       int64   `json:"sales"`
	Stock       int64   `json:"stock"`
	ShopName    string  `json:"shop_name"`
	MainImage   string  `json:"main_image"`
	DetailURL   string  `json:"detail_url"`
	Category    string  `json:"category"`
	CollectedAt string  `json:"collected_at"`
}

// 采集任务配置
type Config struct {
	AppKey        string
	AppSecret     string
	Keywords      []string
	MaxPage       int
	PageSize      int
	MaxConcurrency int
	OutputFile    string
}

func main() {
	// 加载环境变量
	if err := godotenv.Load(); err != nil {
		log.Fatalf("加载环境变量失败: %v", err)
	}

	// 初始化配置
	config := Config{
		AppKey:        os.Getenv("TAOBAO_APP_KEY"),
		AppSecret:     os.Getenv("TAOBAO_APP_SECRET"),
		Keywords:      []string{"手机", "笔记本电脑", "智能手表"}, // 要搜索的关键词
		MaxPage:       5,                                      // 每个关键词最大采集页数
		PageSize:      20,                                     // 每页商品数量
		MaxConcurrency: 10,                                    // 最大并发数
		OutputFile:    "products.json",                        // 输出文件
	}

	// 验证配置
	if config.AppKey == "" || config.AppSecret == "" {
		log.Fatal("请配置TAOBAO_APP_KEY和TAOBAO_APP_SECRET")
	}

	// 创建结果通道
	resultChan := make(chan []Product, config.MaxConcurrency*2)
	
	// 创建上下文，用于控制超时和取消
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()

	// 使用errgroup进行并发控制
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(config.MaxConcurrency)

	// 任务计数器
	var taskCount int
	var taskMutex sync.Mutex

	// 为每个关键词和页面创建采集任务
	for _, keyword := range config.Keywords {
		for page := 1; page <= config.MaxPage; page++ {
			keyword := keyword // 捕获循环变量
			page := page
			
			g.Go(func() error {
				taskMutex.Lock()
				taskCount++
				currentTask := taskCount
				taskMutex.Unlock()
				
				log.Printf("开始采集任务 #%d: 关键词=%s, 页码=%d", currentTask, keyword, page)
				
				// 调用API采集数据
				products, err := fetchProducts(ctx, config, keyword, page)
				if err != nil {
					log.Printf("任务 #%d 失败: %v", currentTask, err)
					return err
				}
				
				log.Printf("任务 #%d 完成: 采集到 %d 个商品", currentTask, len(products))
				resultChan <- products
				return nil
			})
		}
	}

	// 启动一个goroutine等待所有任务完成并关闭通道
	go func() {
		if err := g.Wait(); err != nil {
			log.Printf("部分任务执行失败: %v", err)
		}
		close(resultChan)
	}()

	// 收集所有结果
	var allProducts []Product
	for products := range resultChan {
		allProducts = append(allProducts, products...)
	}

	// 保存结果到文件
	if err := saveResults(allProducts, config.OutputFile); err != nil {
		log.Fatalf("保存结果失败: %v", err)
	}

	log.Printf("所有任务完成，共采集到 %d 个商品，已保存到 %s", len(allProducts), config.OutputFile)
}

// 保存结果到JSON文件
func saveResults(products []Product, filename string) error {
	data, err := json.MarshalIndent(products, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(filename, data, 0644)
}