Golang

8248

Ko kešējāt? Kādas direktorijas?

/* 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 }