Concurrent Crawler с ограничением

Решение

Анализ задачи

Требования:

• Параллельный обход веб-страниц
• Ограничение количества одновременных запросов
• Отслеживание посещённых страниц
• Поддержка отмены через context
• Ограничение по глубине обхода

Ключевые компоненты:

• Semaphore для ограничения конкурентности
• Mutex для синхронизации доступа к visited
• Context для отмены и timeout
• WaitGroup для отслеживания горутин

Решение

import (
    "context"
    "errors"
    "io"
    "net/http"
    "net/url"
    "regexp"
    "strings"
    "sync"
    "time"
)

type Crawler struct {
    maxConcurrent int
    visited       map[string]bool
    mu            sync.Mutex
    client        *http.Client
}

func NewCrawler(maxConcurrent int) *Crawler {
    return &Crawler{
        maxConcurrent: maxConcurrent,
        visited:       make(map[string]bool),
        client: &http.Client{
            Timeout: 5 * time.Second,
        },
    }
}

func (c *Crawler) Crawl(ctx context.Context, startURL string, depth int) ([]string, error) {
    if depth < 0 || startURL == "" {
        return nil, errors.New("invalid parameters")
    }
    
    // Семафор для ограничения конкурентности
    semaphore := make(chan struct{}, c.maxConcurrent)
    
    // WaitGroup для отслеживания горутин
    var wg sync.WaitGroup
    
    // Результаты (все найденные ссылки)
    results := make([]string, 0)
    resultsMu := sync.Mutex{}
    
    // Рекурсивная функция обхода
    var crawlHelper func(u string, d int)
    
    crawlHelper = func(u string, d int) {
        defer wg.Done()
        
        // Проверяем контекст
        select {
        case <-ctx.Done():
            return
        default:
        }
        
        // Проверяем, посещали ли уже эту страницу
        c.mu.Lock()
        if c.visited[u] {
            c.mu.Unlock()
            return
        }
        c.visited[u] = true
        c.mu.Unlock()
        
        // Получаем семафор
        semaphore <- struct{}{}
        defer func() { <-semaphore }()
        
        // Обрабатываем страницу
        links, err := c.fetchLinks(ctx, u)
        if err != nil {
            return
        }
        
        // Добавляем результаты
        resultsMu.Lock()
        results = append(results, links...)
        resultsMu.Unlock()
        
        // Обходим дальше, если есть глубина
        if d > 0 {
            for _, link := range links {
                select {
                case <-ctx.Done():
                    return
                default:
                }
                
                wg.Add(1)
                go crawlHelper(link, d-1)
            }
        }
    }
    
    wg.Add(1)
    go crawlHelper(startURL, depth)
    
    wg.Wait()
    
    return results, nil
}

func (c *Crawler) fetchLinks(ctx context.Context, pageURL string) ([]string, error) {
    req, err := http.NewRequestWithContext(ctx, "GET", pageURL, nil)
    if err != nil {
        return nil, err
    }
    
    resp, err := c.client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    
    if resp.StatusCode != http.StatusOK {
        return nil, errors.New("non-200 status code")
    }
    
    // Читаем HTML
    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return nil, err
    }
    
    // Ищем ссылки
    links := c.extractLinks(string(body), pageURL)
    
    return links, nil
}

func (c *Crawler) extractLinks(html, baseURL string) []string {
    // Простой regex для поиска href
    re := regexp.MustCompile(`href=["']([^"']+)["']`)
    matches := re.FindAllStringSubmatch(html, -1)
    
    var links []string
    baseURLParsed, _ := url.Parse(baseURL)
    
    for _, match := range matches {
        href := match[1]
        
        // Обрабатываем относительные ссылки
        linkURL, _ := url.Parse(href)
        
        if !linkURL.IsAbs() {
            linkURL = baseURLParsed.ResolveReference(linkURL)
        }
        
        // Фильтруем только HTTP(S)
        if linkURL.Scheme == "http" || linkURL.Scheme == "https" {
            links = append(links, linkURL.String())
        }
    }
    
    return links
}

Пояснения реализации

1. Семафор (Channel as Semaphore)

semaphore := make(chan struct{}, c.maxConcurrent)

• Буферизированный канал размером maxConcurrent
• Для входа: semaphore <- struct{}{}
• Для выхода: <-semaphore
• Гарантирует, что одновременно работает максимум N горутин

2. Синхронизация посещённых страниц

• visited map[string]bool с mutex для потокобезопасности
• Проверяем и отмечаем ДО запроса, чтобы избежать дублей

3. Context для отмены

• Проверяем ctx.Done() перед дорогими операциями
• Если context отменён, горутина выходит

4. WaitGroup для отслеживания

• wg.Add() перед запуском горутины
• defer wg.Done() в конце функции
• wg.Wait() ждёт завершения всех

5. Обработка URL

• Разбираем HTML с помощью regex
• Преобразуем относительные ссылки в абсолютные
• Фильтруем только HTTP(S)

Полный пример использования

package main

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

func main() {
    crawler := NewCrawler(3) // Максимум 3 одновременных запроса
    
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    
    links, err := crawler.Crawl(ctx, "https://example.com", 2)
    if err != nil {
        fmt.Println("Error:", err)
        return
    }
    
    fmt.Printf("Found %d links\\n", len(links))
    for _, link := range links {
        fmt.Println("  -", link)
    }
}

Альтернатива: использование golang.org/x/sync/semaphore

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

type Crawler struct {
    maxConcurrent int
    visited       map[string]bool
    mu            sync.Mutex
    client        *http.Client
    sem           *semaphore.Weighted
}

func NewCrawler(maxConcurrent int) *Crawler {
    return &Crawler{
        maxConcurrent: maxConcurrent,
        visited:       make(map[string]bool),
        sem:           semaphore.NewWeighted(int64(maxConcurrent)),
        client:        &http.Client{Timeout: 5 * time.Second},
    }
}

// В crawlHelper:
if err := c.sem.Acquire(ctx, 1); err != nil {
    return
}
defer c.sem.Release(1)

Этот подход более гибкий и стандартный в production коде.

Производительность и опасности

Race Conditions:

• Visited отмечаем под mutex перед запросом
• Results собираем через resultsMu
• Контекст потокобезопасен по определению

Deadlock Prevention:

• Всегда проверяем context перед блокирующими операциями
• Semaphore не может зависнуть, т.к. defer гарантирует освобождение

Утечки горутин:

• WaitGroup гарантирует, что все горутины завершатся
• Семафор предотвращает создание бесконечного количества горутин