saramax

Documentation

Index

• type BatchHandler
  • func NewBatchHandler[T any](l logger.Logger, fn func(msgs []*sarama.ConsumerMessage, t []T) error, ...) *BatchHandler[T]
  • func (h *BatchHandler[T]) Cleanup(session sarama.ConsumerGroupSession) error
  • func (h *BatchHandler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h *BatchHandler[T]) Setup(session sarama.ConsumerGroupSession) error
• type BatchHandlerOption
  • func SetBatchDuration[T any](batchDuration time.Duration) BatchHandlerOption[T]
  • func SetBatchSize[T any](batchSize int) BatchHandlerOption[T]
• type Handler
  • func NewHandler[T any](l logger.Logger, fn func(msg *sarama.ConsumerMessage, t T) error) *Handler[T]
  • func (h *Handler[T]) Cleanup(session sarama.ConsumerGroupSession) error
  • func (h *Handler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (h *Handler[T]) Setup(session sarama.ConsumerGroupSession) error

Types

type BatchHandler

type BatchHandler[T any] struct {
	// contains filtered or unexported fields
}

func NewBatchHandler

func NewBatchHandler[T any](l logger.Logger,
	fn func(msgs []*sarama.ConsumerMessage, t []T) error,
	opts ...BatchHandlerOption[T]) *BatchHandler[T]

NewBatchHandler 实现了ConsumerGroupHandler接口的handler:批量消费后再提交 默认一批 10个 1秒，可用option修改它

func (*BatchHandler[T]) Cleanup

func (h *BatchHandler[T]) Cleanup(session sarama.ConsumerGroupSession) error

func (*BatchHandler[T]) ConsumeClaim

func (h *BatchHandler[T]) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error

ConsumeClaim 批量消费

func (*BatchHandler[T]) Setup

func (h *BatchHandler[T]) Setup(session sarama.ConsumerGroupSession) error

type BatchHandlerOption

type BatchHandlerOption[T any] func(batchHandler *BatchHandler[T])

func SetBatchDuration

func SetBatchDuration[T any](batchDuration time.Duration) BatchHandlerOption[T]

func SetBatchSize

func SetBatchSize[T any](batchSize int) BatchHandlerOption[T]

type Handler

type Handler[T any] struct {
	// contains filtered or unexported fields
}

func NewHandler

func NewHandler[T any](l logger.Logger,
	fn func(msg *sarama.ConsumerMessage, t T) error) *Handler[T]

NewHandler 实现了ConsumerGroupHandler接口的handler:单个消费就提交

func (*Handler[T]) Cleanup

func (h *Handler[T]) Cleanup(session sarama.ConsumerGroupSession) error

func (*Handler[T]) ConsumeClaim

func (h *Handler[T]) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error

func (*Handler[T]) Setup

func (h *Handler[T]) Setup(session sarama.ConsumerGroupSession) error