Senior

/* Mums nepieciešams nodot datus no noteikta avota noteiktam patērētājam. Šajā gadījumā avots sniedz mazākas partijas (~ desmitiem ierakstu), bet patērētājs efektīvāk strādā ar lielākām partijām (~ tūkstošiem ierakstu). Reāls piemērs - datu nodošana no Kafka tipa rindām uz Clickhouse datu bāzi. Avots: - Puslīdz bezgalīgs. - Avots nekad neatgriež vairāk par MaxItems ierakstiem vienā Next izsaukumā. - Vienā "sesijā" (viena Pipe funkcijas izsaukuma laikā) avots katru reizi atgriež jaunus datus. - Tomēr, pārstartējot, avots atsāks no iepriekšējās "apstiprinātās" pozīcijas, kas norādīta ar cookie. Tādēļ katra vērtība, ko atgrieza Next izsaukums, pēc datu saglabāšanas saņēmējam, ir jāapstiprina ar Commit izsaukumu, un tas jāveic tieši tajā pašā secībā, kurā tie tika atgriezti ar Next. Saņēmējs: - Nevar apstrādāt vairāk par MaxItems vienlaikus. Galvenais līmenis: Jāievieš funkcija func Pipe(p Producer, c Consumer) error, kura lasa datus no avota, tos grupē buferī, kura izmērs nav lielāks par MaxItems, un saglabā saņēmējā, tad atjauno progresu avotā. Grūtības: Next, Process un Commit metodes ir saistītas ar tīkla izsaukumiem un var darboties diezgan ilgi. Lai paātrinātu datu pārraidi, nepieciešams paralēli veikt lasīšanu, ierakstīšanu un apstiprināšanu. Tādējādi, izsaucot Process vai Commit, turpinās lasīšana no avota un jauna bufera veidošana. */

sobes.tech AI

Atbilde no AI

Для реализации функции Pipe с параллельной обработкой чтения, записи и подтверждения прогресса можно использовать каналы и горутины. Идея:

Одна горутина читает данные из источника (Producer.Next), формируя батчи не более MaxItems.
Вторая горутина обрабатывает батчи (Consumer.Process).
Третья горутина подтверждает прогресс (Producer.Commit) строго в порядке получения cookie.

Для сохранения порядка подтверждения можно использовать очередь или канал с буфером, где cookie будут храниться в порядке поступления.

Пример реализации:

func Pipe(p Producer, c Consumer) error {
    type batch struct {
        items  []any
        cookie int
    }

    batches := make(chan batch, 10)      // Канал для передачи батчей на обработку
    commitCh := make(chan int, 10)       // Канал для передачи cookie на подтверждение
    errCh := make(chan error, 3)         // Канал для ошибок
    done := make(chan struct{})           // Канал для остановки

    // Чтение данных
    go func() {
        defer close(batches)
        for {
            items, cookie, err := p.Next()
            if err != nil {
                errCh <- err
                return
            }
            if len(items) == 0 {
                // Источник пуст, можно завершать
                return
            }

            // Формируем батчи не более MaxItems
            for start := 0; start < len(items); start += MaxItems {
                end := start + MaxItems
                if end > len(items) {
                    end = len(items)
                }
                batches <- batch{items: items[start:end], cookie: cookie}
            }
        }
    }()

    // Обработка данных
    go func() {
        defer close(commitCh)
        for b := range batches {
            if err := c.Process(b.items); err != nil {
                errCh <- err
                return
            }
            commitCh <- b.cookie
        }
    }()

    // Подтверждение
    go func() {
        for cookie := range commitCh {
            if err := p.Commit(cookie); err != nil {
                errCh <- err
                return
            }
        }
        close(done)
    }()

    select {
    case err := <-errCh:
        return err
    case <-done:
        return nil
    }
}

В этом примере:

Чтение и формирование батчей происходит в первой горутине.
Обработка батчей — во второй.
Подтверждение cookie — в третьей.

Каналы обеспечивают поток данных и порядок подтверждения. Ошибки передаются через errCh, что позволяет прервать работу при возникновении ошибки.

Такой подход позволяет параллельно читать, обрабатывать и подтверждать данные, повышая общую производительность.

Iepriekšējais Tālāk