alikafkamq

Documentation

Index

• Constants
• func WithGroupName(gpn string) consumerOpt
• func WithName(n string) consumerOpt
• func WithRetry(n int32) consumerOpt
• type Consumer
  • func NewConsumer(cred *KafkaMQCredentials, ots *OtsConfig, rds *RedisConfig, ...) (*Consumer, error)
  • func (c *Consumer) Cleanup(session sarama.ConsumerGroupSession) error
  • func (c *Consumer) Consume(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim, ...) error
  • func (c *Consumer) Setup(session sarama.ConsumerGroupSession) error
  • func (c *Consumer) Start() error
  • func (c *Consumer) Stop() error
• type Handler
• type KafkaMQCredentials
• type M
• type OtsConfig
• type RedisConfig
• type Subscription

Constants

const (
	XServiceChainReal    = "real"
	XServiceChainVirtual = "virtual"
)

const (
	KAFKAENV = "kakfaDebug"
)

Functions

func WithGroupName

func WithGroupName(gpn string) consumerOpt

func WithName

func WithName(n string) consumerOpt

func WithRetry

func WithRetry(n int32) consumerOpt

Types

type Consumer

type Consumer struct {
	Name string

	GroupName string
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(cred *KafkaMQCredentials, ots *OtsConfig, rds *RedisConfig, subs []Subscription, opts ...consumerOpt) (*Consumer, error)

初始化

func (*Consumer) Cleanup

func (c *Consumer) Cleanup(session sarama.ConsumerGroupSession) error

func (*Consumer) Consume

func (c *Consumer) Consume(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim, msg *sarama.ConsumerMessage) error

func (*Consumer) Setup

func (c *Consumer) Setup(session sarama.ConsumerGroupSession) error

func (*Consumer) Start

func (c *Consumer) Start() error

启动

func (*Consumer) Stop

func (c *Consumer) Stop() error

停止

type Handler

type Handler func(*M) error

type KafkaMQCredentials

type KafkaMQCredentials struct {
	AppName   string
	Endpoints []string
	User      string
	Password  string
}

type M

type M struct {
	MsgId           string
	Topic           string
	Tag             string
	Key             string
	Body            []byte
	ReConsumerTimes int32
}

type OtsConfig

type OtsConfig struct {
	EndPoint        string `json:"end_point"`
	InstanceName    string `json:"instance_name"`
	AccessKeyID     string `json:"access_key_id"`
	AccessKeySecret string `json:"access_key_secret"`
}

type RedisConfig

type RedisConfig struct {
	DB       int    `json:"db"`
	Address  string `json:"address"`
	Password string `json:"password"`
	PoolSize int    `json:"pool_size"`
}

type Subscription

type Subscription struct {
	Topic   string
	Tag     string
	Key     string
	Partion int32
	Offset  int32
	Fn      []Handler
}