Middle

/* Mums nepieciešams pārsūtīt datus no noteikta avota noteiktam patērētājam. Šajā procesā avots sniedz datus mazās partijās (~ desmitiem ierakstu), bet patērētājs efektīvāk strādā ar lielākiem batčiem. Reāls piemērs - datu piegāde no Kafka tipa rindām uz Clickhouse datubāzi. Avots: - Nosacīti bezgalīgs. - Avots nekad neatgriezīs vairāk par MaxItems ierakstiem vienā Next izsaukumā. - Vienā "sesijā" (viena Pipe funkcijas izsaukuma laikā) avots katru reizi atgriež jaunus datus. - Tomēr, pēc pārstartēšanas, avots sāk no iepriekšējās "apstiprinātās" pozīcijas, kas norādīta ar cookie. Tādēļ, katra cookie vērtība, ko atgrieza Next, pēc datu saglabāšanas saņēmējā, jāapstiprina ar Commit, un tas jāizdara stingri tādā pašā secībā, kā tie tika atgriezti. Saņēmējs: - Nevar apstrādāt vairāk par MaxItems vienlaikus. Pamatlīmenis: Jāimplementē funkcija func Pipe(p Producer, c Consumer) error, kura lasa datus no avota, tos grupē buferī, kura lielums nepārsniedz MaxItems, un saglabā saņēmējā, pēc tam apstiprina progresu avotā. */ const MaxItems = 9999 type Producer interface { // Next atgriež: // - datu partiju // - cookie apstiprināšanai // - kļūdu Next() (items []any, cookie int, err error) // Commit apzīmē, ka datu partija ir apstrādāta Commit(cookie int) error } type Consumer interface { Process(items []any) error } func Pipe(p Producer, c Consumer) error { var buf []any var cookies []int for { items, cookie, err := p.Next() if err != nil { return err } buf = append(buf, items...) cookies = append(cookies, cookie) if len(buf) >= MaxItems { if err := c.Process(buf); err != nil { return err } for _, c := range cookies { if err := p.Commit(c); err != nil { return err } } buf = buf[:0] cookies = nil } } if len(buf) > 0 { if err := c.Process(buf); err != nil { return err } for _, c := range cookies { if err := p.Commit(c); err != nil { return err } } } return nil }

sobes.tech AI

Atbilde no AI

Ваша задача — реализовать функцию Pipe, которая читает данные из источника (Producer) небольшими пачками, буферизует их до размера не более MaxItems, передает в Consumer и затем фиксирует прогресс в Producer через Commit в правильном порядке.

Основные моменты:

Источник возвращает данные пачками, максимум MaxItems за вызов Next.
Нужно аккумулировать данные в буфере, не превышая MaxItems.
После передачи данных в Consumer вызываем Commit для всех cookie в порядке получения.
Источник бесконечен, поэтому цикл бесконечный.

В представленном коде есть ошибка: бесконечный цикл не прерывается, и код после цикла недостижим. Нужно предусмотреть условие выхода (например, когда Next возвращает пустой срез и нет ошибки).

Исправленный пример реализации:

func Pipe(p Producer, c Consumer) error {
    var buf []any
    var cookies []int

    for {
        items, cookie, err := p.Next()
        if err != nil {
            return err
        }
        // Если источник вернул пустой срез, считаем, что данных больше нет
        if len(items) == 0 {
            break
        }

        buf = append(buf, items...)
        cookies = append(cookies, cookie)

        // Если буфер достиг максимума, отправляем на обработку
        if len(buf) >= MaxItems {
            if err := c.Process(buf); err != nil {
                return err
            }
            for _, c := range cookies {
                if err := p.Commit(c); err != nil {
                    return err
                }
            }
            buf = buf[:0]
            cookies = nil
        }
    }

    // Обработка оставшихся данных
    if len(buf) > 0 {
        if err := c.Process(buf); err != nil {
            return err
        }
        for _, c := range cookies {
            if err := p.Commit(c); err != nil {
                return err
            }
        }
    }

    return nil
}

Таким образом, мы аккумулируем данные, не превышая MaxItems, и после обработки фиксируем cookie в порядке получения, что соответствует требованиям.

Iepriekšējais Tālāk