kafka

package

v0.0.0-...-fd7de33 Latest Latest Go to latest Published: May 8, 2024 License: Apache-2.0 Imports: 17 Imported by: 4

Details

View Source

const (
	SpecifiedPartitionKey = "specifiedPartition"
)

View Source

var ErrInvalidArgument = errors.New("invalid argument")

func NewMQ(network string, address []string) (mq.MQ, error)

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

func NewConsumer(address []string, topic, groupID string) *Consumer

func (c *Consumer) Close() error

func (c *Consumer) Consume(ctx context.Context) (*mq.Message, error)

func (c *Consumer) ConsumeChan(ctx context.Context) (<-chan *mq.Message, error)

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

func (m *MQ) Close() error

func (m *MQ) Consumer(topic, groupID string) (mq.Consumer, error)

func (m *MQ) CreateTopic(ctx context.Context, name string, partitions int) error

func (m *MQ) DeleteTopics(ctx context.Context, topics ...string) error

DeleteTopics 删除topic

func (m *MQ) Producer(topic string) (mq.Producer, error)

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

func NewProducer(address []string, topic string, balancer kafkago.Balancer) *Producer

func (p *Producer) Close() error

func (p *Producer) Produce(ctx context.Context, m *mq.Message) (*mq.ProducerResult, error)

func (p *Producer) ProduceWithPartition(ctx context.Context, m *mq.Message, partition int) (*mq.ProducerResult, error)

ProduceWithPartition 并没有校验 partition 的正确性。

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

SpecifiedPartitionBalancer 借助kafka客户端提供的Balancer接口实现向指定分区生产消息如果在message.WriterData中找到指定的partition信息则直接用其作为目标partition 如果没有找到则使用默认负载均衡器计算目标partition

func NewSpecifiedPartitionBalancer(defaultBalancer kafkago.Balancer) (*SpecifiedPartitionBalancer, error)

func (b *SpecifiedPartitionBalancer) Balance(msg kafkago.Message, partitions ...int) (partition int)

Path	Synopsis
common