kafka

package

v0.0.0-...-b7e27bc Latest Latest Go to latest Published: Apr 6, 2024 License: Apache-2.0 Imports: 17 Imported by: 1

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

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