type Batcher struct {
  opts options
  Do       func(ctx context.Context, val map[string][]interface{})
  Sharding func(key string) int
  chans    []chan *msg
  wait     sync.WaitGroup
}

type msg struct {
  key string
  val interface{}
}
type Batcher struct {
  opts options
  Do       func(ctx context.Context, val map[string][]interface{})
  Sharding func(key string) int
  chans    []chan *msg
  wait     sync.WaitGroup
}
func New(opts ...Option) *Batcher {
  b := &Batcher{}
  for _, opt := range opts {
    opt.apply(&b.opts)
  }
  b.opts.check()
  b.chans = make([]chan *msg, b.opts.worker)
  for i := 0; i < b.opts.worker; i++ {
    b.chans[i] = make(chan *msg, b.opts.buffer)
  }
  return b
}
func (b *Batcher) Start() {
  if b.Do == nil {
    log.Fatal("Batcher: Do func is nil")
  }
  if b.Sharding == nil {
    log.Fatal("Batcher: Sharding func is nil")
  }
  b.wait.Add(len(b.chans))
  for i, ch := range b.chans {
    go b.merge(i, ch)
  }
}
func (b *Batcher) Add(key string, val interface{}) error {
  ch, msg := b.add(key, val)
  select {
  case ch <- msg:
  default:
    return ErrFull
  }
  return nil
}
func (b *Batcher) add(key string, val interface{}) (chan *msg, *msg) {
  sharding := b.Sharding(key) % b.opts.worker
  ch := b.chans[sharding]
  msg := &msg{key: key, val: val}
  return ch, msg
}
func (b *Batcher) merge(idx int, ch <-chan *msg) {
  defer b.wait.Done()
  var (
    msg        *msg
    count      int
    closed     bool
    lastTicker = true
    interval   = b.opts.interval
    vals       = make(map[string][]interface{}, b.opts.size)
  )
  if idx > 0 {
    interval = time.Duration(int64(idx) * (int64(b.opts.interval) / int64(b.opts.worker)))
  }
  ticker := time.NewTicker(interval)
  for {
    select {
    case msg = <-ch:
      if msg == nil {
        closed = true
        break
      }
      count++
      vals[msg.key] = append(vals[msg.key], msg.val)
      if count >= b.opts.size {
        break
      }
      continue
    case <-ticker.C:
      if lastTicker {
        ticker.Stop()
        ticker = time.NewTicker(b.opts.interval)
        lastTicker = false
      }
    }
    if len(vals) > 0 {
      ctx := context.Background()
      b.Do(ctx, vals)
      vals = make(map[string][]interface{}, b.opts.size)
      count = 0
    }
    if closed {
      ticker.Stop()
      return
    }
  }
}
func (b *Batcher) Close() {
  for _, ch := range b.chans {
    ch <- nil
  }
  b.wait.Wait()
}

b := batcher.New(
  batcher.WithSize(batcherSize),
  batcher.WithBuffer(batcherBuffer),
  batcher.WithWorker(batcherWorker),
  batcher.WithInterval(batcherInterval),
)
b.Sharding = func(key string) int {
  pid, _ := strconv.ParseInt(key, 10, 64)
  return int(pid) % batcherWorker
}
b.Do = func(ctx context.Context, val map[string][]interface{}) {
  var msgs []*KafkaData
  for _, vs := range val {
    for _, v := range vs {
      msgs = append(msgs, v.(*KafkaData))
    }
  }
  kd, err := json.Marshal(msgs)
  if err != nil {
    logx.Errorf("Batcher.Do json.Marshal msgs: %v error: %v", msgs, err)
  }
  if err = s.svcCtx.KafkaPusher.Push(string(kd)); err != nil {
    logx.Errorf("KafkaPusher.Push kd: %s error: %v", string(kd), err)
  }
}
s.batcher = b
s.batcher.Start()

err = l.batcher.Add(strconv.FormatInt(in.ProductId, 10), &KafkaData{Uid: in.UserId, Pid: in.ProductId})
if err!= nil {
    logx.Errorf("l.batcher.Add uid: %d pid: %d error: %v", in.UserId, in.ProductId, err)
}

func NewService(c config.Config) *Service {
  s := &Service{
    c:          c,
    ProductRPC: product.NewProduct(zrpc.MustNewClient(c.ProductRPC)),
    OrderRPC:   order.NewOrder(zrpc.MustNewClient(c.OrderRPC)),
    msgsChan:   make([]chan *KafkaData, chanCount),
  }
  for i := 0; i < chanCount; i++ {
    ch := make(chan *KafkaData, bufferCount)
    s.msgsChan[i] = ch
    s.waiter.Add(1)
    go s.consume(ch)
  }
  return s
}

func (s *Service) Consume(_ string, value string) error {
  logx.Infof("Consume value: %s\n", value)
  var data []*KafkaData
  if err := json.Unmarshal([]byte(value), &data); err != nil {
    return err
  }
  for _, d := range data {
    s.msgsChan[d.Pid%chanCount] <- d
  }
  return nil
}

func (s *Service) consume(ch chan *KafkaData) {
  defer s.waiter.Done()
  for {
    m, ok := <-ch
    if !ok {
      log.Fatal("seckill rmq exit")
    }
    fmt.Printf("consume msg: %+v\n", m)
    p, err := s.ProductRPC.Product(context.Background(), &product.ProductItemRequest{ProductId: m.Pid})
    if err != nil {
      logx.Errorf("s.ProductRPC.Product pid: %d error: %v", m.Pid, err)
      return
    }
    if p.Stock <= 0 {
      logx.Errorf("stock is zero pid: %d", m.Pid)
      return
    }
    _, err = s.OrderRPC.CreateOrder(context.Background(), &order.CreateOrderRequest{Uid: m.Uid, Pid: m.Pid})
    if err != nil {
      logx.Errorf("CreateOrder uid: %d pid: %d error: %v", m.Uid, m.Pid, err)
      return
    }
    _, err = s.ProductRPC.UpdateProductStock(context.Background(), &product.UpdateProductStockRequest{ProductId: m.Pid, Num: 1})
    if err != nil {
      logx.Errorf("UpdateProductStock uid: %d pid: %d error: %v", m.Uid, m.Pid, err)
    }
  }
}

const (
  luaCheckAndUpdateScript = `
local counts = redis.call("HMGET", KEYS[1], "total", "seckill")
local total = tonumber(counts[1])
local seckill = tonumber(counts[2])
if seckill + 1 <= total then
  redis.call("HINCRBY", KEYS[1], "seckill", 1)
  return 1
end
return 0
`
)
func (l *CheckAndUpdateStockLogic) CheckAndUpdateStock(in *product.CheckAndUpdateStockRequest) (*product.CheckAndUpdateStockResponse, error) {
  val, err := l.svcCtx.BizRedis.EvalCtx(l.ctx, luaCheckAndUpdateScript, []string{stockKey(in.ProductId)})
  if err != nil {
    return nil, err
  }
  if val.(int64) == 0 {
    return nil, status.Errorf(codes.ResourceExhausted, fmt.Sprintf("insufficient stock: %d", in.ProductId))
  }
  return &product.CheckAndUpdateStockResponse{}, nil
}
func stockKey(pid int64) string {
  return fmt.Sprintf("stock:%d", pid)
}

func (s *Service) consume(ch chan *KafkaData) {
  defer s.waiter.Done()
  for {
    m, ok := <-ch
    if !ok {
      log.Fatal("seckill rmq exit")
    }
    fmt.Printf("consume msg: %+v\n", m)
    _, err := s.ProductRPC.CheckAndUpdateStock(context.Background(), &product.CheckAndUpdateStockRequest{ProductId: m.Pid})
    if err != nil {
      logx.Errorf("s.ProductRPC.CheckAndUpdateStock pid: %d error: %v", m.Pid, err)
      return
    }
    _, err = s.OrderRPC.CreateOrder(context.Background(), &order.CreateOrderRequest{Uid: m.Uid, Pid: m.Pid})
    if err != nil {
      logx.Errorf("CreateOrder uid: %d pid: %d error: %v", m.Uid, m.Pid, err)
      return
    }
    _, err = s.ProductRPC.UpdateProductStock(context.Background(), &product.UpdateProductStockRequest{ProductId: m.Pid, Num: 1})
    if err != nil {
      logx.Errorf("UpdateProductStock uid: %d pid: %d error: %v", m.Uid, m.Pid, err)
    }
  }
}

cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
if err != nil {
   log.Fatal(err)
}
defer cli.Close()
session, err := concurrency.NewSession(cli, concurrency.WithTTL(10))
if err != nil {
   log.Fatal(err)
}
defer session.Close()
mux := concurrency.NewMutex(session, "lock")
if err := mux.Lock(context.Background()); err != nil {
   log.Fatal(err)
}
if err := mux.Unlock(context.Background()); err != nil {
   log.Fatal(err)
}