Golang

8247

Ne önbelleğe alındı? Bazı dizinler mi?

/* Bir kaynaktan bir tüketiciye veri aktarmamız gerekiyor. Kaynak, verileri küçük paketler halinde (~on kayıt) verirken, tüketici daha büyük paketlerle daha verimli çalışır. Gerçek bir örnek, Kafka türü kuyruklardan Clickhouse veritabanına veri aktarımıdır. Kaynak: - Neredeyse sonsuz. - Kaynak, bir Next çağrısında hiçbir zaman MaxItems'ten fazla kayıt döndürmez. - Bir "oturum" (bir Pipe fonksiyon çağrısı) içinde, kaynak her Next çağrısında yeni veriler döndürür. - Ancak, yeniden başlatıldıktan sonra, kaynak önceki "onaylanmış" konumdan, cookie ile belirlenen noktadan başlar. Bu nedenle, Next'in döndürdüğü her cookie değeri, veriler alıcıya kaydedildikten sonra, aynı sırayla Commit çağrısı ile onaylanmalıdır. Alıcı: - Aynı anda MaxItems'ten fazla işleyemez. Temel seviye: Gerekli olan, func Pipe(p Producer, c Consumer) error fonksiyonunu uygulamaktır, bu fonksiyon kaynaktan veri okur, bunları MaxItems boyutunda bir tamponda gruplar ve alıcıya kaydeder, ve ardından kaynaktaki ilerlemeyi onaylar. */ const MaxItems = 9999 type Producer interface { // Next,: // - işlenmek üzere öğe paketleri döner // - işleme tamamlandığında onaylanacak cookie // - hata Next() (items []any, cookie int, err error) // Commit, işlenmiş veri paketini işaretlemek için kullanılır 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 }