subscriber

Documentation

Index

• func Close()
• func InitPushConsumerSubscribes(opts map[string]*RmqPushConsumerOptions, ...)
• type IRocketMQConsumerSubscribeHandler
• type RmqPushConsumerOptions
  • func DefaultRmqPushConsumerOptions() *RmqPushConsumerOptions

Functions

func Close

func Close()

func InitPushConsumerSubscribes

func InitPushConsumerSubscribes(opts map[string]*RmqPushConsumerOptions, mapSubscribeHandler map[string]IRocketMQConsumerSubscribeHandler)

InitPushConsumerSubscribes support msg tag subscribe

Types

type IRocketMQConsumerSubscribeHandler

type IRocketMQConsumerSubscribeHandler interface {

	// SubMsgsHandle sub some msg to consume
	// maybe u can concurency batch do to improve throughput rate,
	// with min(ConsumeMessageBatchMaxSize,PullBatchSize)>1;
	// but batch done to return commit status
	// so pelease Idempotent consume every msg
	// notice: PullBatchSize/min(ConsumeMessageBatchMaxSize,PullBatchSize) subscribes are concurency
	SubMsgsHandle(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error)
}

type RmqPushConsumerOptions

type RmqPushConsumerOptions struct {
	Name            string   `mapstructure:"name"`
	NameSrvs        []string `mapstructure:"nameSrvs"`
	GroupName       string   `mapstructure:"groupName"`
	TopicName       string   `mapstructure:"topicName"`
	Tag             string   `mapstructure:"tag"`
	PullRetryCn     int      `mapstructure:"pullRetryCn"`
	LogicMaxRetryCn int      `mapstructure:"logicMaxRetryCn"`

	// MaxReconsumeTimes retry over to dlq topic, -1 is default 16, 0 or <-1 don't retry,over to dlq topic
	MaxReconsumeTimes int `mapstructure:"maxReconsumeTimes"`

	PullBatchSize              int `mapstructure:"pullBatchSize"`
	ConsumeMessageBatchMaxSize int `mapstructure:"consumeMessageBatchMaxSize"`

	// The DelayLevel specify the waiting time that before next reconsume,
	// and this range is from 1 to 18 now.
	// The time of each level is the value of indexing of {level-1} in
	// [1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h]
	//delayLevel := 2
	// out delay level range, <0 don't retry, retry cn: maxReconsumeTimes default 16
	// [10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h]
	DelayLevel int `mapstructure:"delayLevel"`
}

func DefaultRmqPushConsumerOptions

func DefaultRmqPushConsumerOptions() *RmqPushConsumerOptions