article

Documentation

Index

• type HistoryReadEventConsumer
  • func NewHistoryReadEventConsumer(client sarama.Client, l logger.LoggerV1, repo repository.InteractiveRepository) *HistoryReadEventConsumer
  • func (r *HistoryReadEventConsumer) Consume(msg *sarama.ConsumerMessage, t ReadEvent) error
  • func (r *HistoryReadEventConsumer) Start() error
• type InteractiveReadEventBatchConsumer
  • func NewInteractiveReadEventBatchConsumer(client sarama.Client, repo repository.InteractiveRepository, l logger.LoggerV1) *InteractiveReadEventBatchConsumer
  • func (r *InteractiveReadEventBatchConsumer) Consume(msg []*sarama.ConsumerMessage, ts []ReadEvent) error
  • func (r *InteractiveReadEventBatchConsumer) Start() error
• type InteractiveReadEventConsumer
  • func NewInteractiveReadEventConsumer(client sarama.Client, l logger.LoggerV1, repo repository.InteractiveRepository) *InteractiveReadEventConsumer
  • func (r *InteractiveReadEventConsumer) Consume(msg *sarama.ConsumerMessage, t ReadEvent) error
  • func (r *InteractiveReadEventConsumer) Start() error
• type KafkaProducer
  • func (k *KafkaProducer) ProduceReadEvent(ctx context.Context, evt ReadEvent) error
  • func (k *KafkaProducer) ProduceReadEventV1(ctx context.Context, v1 ReadEventV1)
• type Producer
  • func NewKafkaProducer(pc sarama.SyncProducer) Producer
• type ReadEvent
• type ReadEventV1

Types

type HistoryReadEventConsumer

type HistoryReadEventConsumer struct {
	// contains filtered or unexported fields
}

func NewHistoryReadEventConsumer

func NewHistoryReadEventConsumer(
	client sarama.Client,
	l logger.LoggerV1,
	repo repository.InteractiveRepository) *HistoryReadEventConsumer

func (*HistoryReadEventConsumer) Consume

func (r *HistoryReadEventConsumer) Consume(msg *sarama.ConsumerMessage, t ReadEvent) error

Consume 这个不是幂等的

func (*HistoryReadEventConsumer) Start

func (r *HistoryReadEventConsumer) Start() error

type InteractiveReadEventBatchConsumer

type InteractiveReadEventBatchConsumer struct {
	// contains filtered or unexported fields
}

func NewInteractiveReadEventBatchConsumer

func NewInteractiveReadEventBatchConsumer(client sarama.Client, repo repository.InteractiveRepository, l logger.LoggerV1) *InteractiveReadEventBatchConsumer

func (*InteractiveReadEventBatchConsumer) Consume

func (r *InteractiveReadEventBatchConsumer) Consume(msg []*sarama.ConsumerMessage, ts []ReadEvent) error

Consume 这个不是幂等的

func (*InteractiveReadEventBatchConsumer) Start

func (r *InteractiveReadEventBatchConsumer) Start() error

type InteractiveReadEventConsumer

type InteractiveReadEventConsumer struct {
	// contains filtered or unexported fields
}

func NewInteractiveReadEventConsumer

func NewInteractiveReadEventConsumer(
	client sarama.Client,
	l logger.LoggerV1,
	repo repository.InteractiveRepository) *InteractiveReadEventConsumer

func (*InteractiveReadEventConsumer) Consume

func (r *InteractiveReadEventConsumer) Consume(msg *sarama.ConsumerMessage, t ReadEvent) error

Consume 这个不是幂等的

func (*InteractiveReadEventConsumer) Start

func (r *InteractiveReadEventConsumer) Start() error

type KafkaProducer

type KafkaProducer struct {
	// contains filtered or unexported fields
}

func (*KafkaProducer) ProduceReadEvent

func (k *KafkaProducer) ProduceReadEvent(ctx context.Context, evt ReadEvent) error

ProduceReadEvent 如果你有复杂的重试逻辑，就用装饰器 你认为你的重试逻辑很简单，你就放这里

func (*KafkaProducer) ProduceReadEventV1

func (k *KafkaProducer) ProduceReadEventV1(ctx context.Context, v1 ReadEventV1)

type Producer

type Producer interface {
	ProduceReadEvent(ctx context.Context, evt ReadEvent) error
	ProduceReadEventV1(ctx context.Context, v1 ReadEventV1)
}

func NewKafkaProducer

func NewKafkaProducer(pc sarama.SyncProducer) Producer

type ReadEvent

type ReadEvent struct {
	Uid int64
	Aid int64
}

type ReadEventV1

type ReadEventV1 struct {
	Uids []int64
	Aids []int64
}