func (b *BulkConsumer) Consume(batchSize int, interval int, ctx context.Context, wg *sync.WaitGroup) {
	chunk := make([]*params, batchSize)
	i := 0
	timer := time.NewTimer(time.Duration(interval) * time.Second)
	commit := func() {
		defer func() {
			i = 0
		}()

		if i == 0 {
			return
		}
		log.Debug("es update body: " + getUpdateStr(chunk, i))
                 // elasticsearch bulk 
		res, err := b.c.Bulk(strings.NewReader(getUpdateStr(chunk, i)))
		log.Debug(res, err)
		if err != nil || res.StatusCode >= 300 {
			log.Errorf("es bulk fail. res: %s, error: %s", res, err)
		}

		pipe := b.r.Pipeline()
		for j := 0; j < i; j++ {
			pipe.SAdd("member_refresh_tag", chunk[j].memberId)
		}
		_, err = pipe.Exec()
		if err != nil {
			log.Errorf("set member_refresh_tag cache error. error: %s", err)
		}
	}

	for {
		select {
		case <-timer.C:
			commit()
			timer.Reset(time.Duration(interval) * time.Second)
		case p := <-b.ch:
			chunk[i] = &p
			i++
			if i >= batchSize {
				if !timer.Stop() {
					<-timer.C
				}
				commit()
				timer.Reset(time.Duration(interval) * time.Second)
			}
		case <-ctx.Done():
			log.Info("ctx.Done()")
			if !timer.Stop() {
				<-timer.C
			}
			log.Info("bulk_consumer exiting")
			commit()
			log.Info("bulk_consumer exit")
			wg.Done()
			return
		}
	}
}

package main

import (
	"github.com/elastic/go-elasticsearch/v7"
	"strings"
	"time"
)

func getUpdateStr() string {
	return strings.Join([]string{
		"{\"update\": { \"_id\":    33141, \"_index\": \"member_tag_0\" }}",
		`{"doc": { "channel_uid": 94248 }}`,
	}, "")
}

func main() {
	es, _ := elasticsearch.NewDefaultClient()

	for {
		_, _ = es.Bulk(strings.NewReader(getUpdateStr()))
		time.Sleep(100 * time.Millisecond)
	}
}