kafka

package module
v0.0.0-...-6d4b899 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 12, 2024 License: Apache-2.0 Imports: 6 Imported by: 0

README

golang kafka tool

kafka 常用命令

【文献】 消费者组天剑消费者：https://www.jianshu.com/p/df17c003f180 https://segmentfault.com/a/1190000040820548 https://zhuanlan.zhihu.com/p/485983874 https://www.cnblogs.com/wishFreedom/p/15131600.html https://blog.csdn.net/luslin1711/article/details/105798571 https://blog.csdn.net/qq_38020553/article/details/103609540 https://www.cnblogs.com/HappyTeemo/p/17262261.html https://segmentfault.com/a/1190000041236865

Documentation

Overview

Package kafka Date: 2023/11/30 15:02 Author: Amu Description:

Package kafka Date: 2023/11/30 15:03 Author: Amu Description:

Package kafka Date: 2023/11/30 15:05 Author: Amu Description:

Package kafka Date: 2023/11/24 17:35 Author: Amu Description:

Package kafka Date: 2023/11/30 15:01:29 Author: Amu Description:

Index

Constants

View Source 
const (
	OffsetNewest = sarama.OffsetNewest
	OffsetOldest = sarama.OffsetOldest
	WaitNone     = sarama.NoResponse
	WaitLeader   = sarama.WaitForLocal
	WaitAll      = sarama.WaitForAll
)

Variables

View Source 
var (
	V0_8_2_0  = sarama.V0_8_2_0
	V0_8_2_1  = sarama.V0_8_2_1
	V0_8_2_2  = sarama.V0_8_2_2
	V0_9_0_0  = sarama.V0_9_0_0
	V0_9_0_1  = sarama.V0_9_0_1
	V0_10_0_0 = sarama.V0_10_0_0
	V0_10_0_1 = sarama.V0_10_0_1
	V0_10_1_0 = sarama.V0_10_1_0
	V0_10_1_1 = sarama.V0_10_0_1
	V0_10_2_0 = sarama.V0_10_2_0
	V0_10_2_1 = sarama.V0_10_2_1
	V0_10_2_2 = sarama.V0_10_2_2
	V0_11_0_0 = sarama.V0_11_0_0
	V0_11_0_1 = sarama.V0_11_0_1
	V0_11_0_2 = sarama.V0_11_0_2
	V1_0_0_0  = sarama.V1_0_0_0
	V1_0_1_0  = sarama.V1_0_1_0
	V1_0_2_0  = sarama.V1_0_2_0
	V1_1_0_0  = sarama.V1_1_0_0
	V1_1_1_0  = sarama.V1_1_1_0
	V2_0_0_0  = sarama.V2_0_0_0
	V2_0_1_0  = sarama.V2_0_1_0
	V2_1_0_0  = sarama.V2_1_0_0
	V2_1_1_0  = sarama.V2_1_1_0
	V2_2_0_0  = sarama.V2_2_0_0
	V2_2_1_0  = sarama.V2_2_1_0
	V2_2_2_0  = sarama.V2_2_2_0
	V2_3_0_0  = sarama.V2_3_0_0
	V2_3_1_0  = sarama.V2_3_1_0
	V2_4_0_0  = sarama.V2_4_0_0
	V2_4_1_0  = sarama.V2_4_1_0
	V2_5_0_0  = sarama.V2_5_0_0
	V2_5_1_0  = sarama.V2_5_1_0
	V2_6_0_0  = sarama.V2_6_0_0
	V2_6_1_0  = sarama.V2_6_1_0
	V2_6_2_0  = sarama.V2_6_2_0
	V2_7_0_0  = sarama.V2_7_0_0
	V2_7_1_0  = sarama.V2_7_1_0
	V2_8_0_0  = sarama.V2_8_0_0
	V2_8_1_0  = sarama.V2_8_1_0
	V2_8_2_0  = sarama.V2_8_2_0
	V3_0_0_0  = sarama.V3_0_0_0
	V3_0_1_0  = sarama.V3_0_1_0
	V3_0_2_0  = sarama.V3_0_2_0
	V3_1_0_0  = sarama.V3_1_0_0
	V3_1_1_0  = sarama.V3_1_1_0
	V3_1_2_0  = sarama.V3_1_2_0
	V3_2_0_0  = sarama.V3_2_0_0
	V3_2_1_0  = sarama.V3_2_1_0
	V3_2_2_0  = sarama.V3_2_2_0
	V3_2_3_0  = sarama.V3_2_3_0
	V3_3_0_0  = sarama.V3_3_0_0
	V3_3_1_0  = sarama.V3_3_1_0
	V3_3_2_0  = sarama.V3_3_2_0
	V3_4_0_0  = sarama.V3_4_0_0
	V3_4_1_0  = sarama.V3_4_1_0
	V3_5_0_0  = sarama.V3_5_0_0
	V3_5_1_0  = sarama.V3_5_1_0
	V3_6_0_0  = sarama.V3_6_0_0
)

Functions

This section is empty.

Types

type ByteEncoder 

type ByteEncoder []byte

func (ByteEncoder) Encode 

func (b ByteEncoder) Encode() ([]byte, error)

func (ByteEncoder) Length 

func (b ByteEncoder) Length() int

type Consumer 

type Consumer interface {
	GetMsg() *sarama.ConsumerMessage
	Submit()
}

type ConsumerMessage 

type ConsumerMessage *sarama.ConsumerMessage

type Manager 

type Manager struct {
	Producer         sarama.AsyncProducer
	ConsumerGroup    sarama.ConsumerGroup
	ConsumerMessages chan ConsumerMessage
	Consumers        chan Consumer
	// contains filtered or unexported fields
}

func New 

func New(opts ...Option) (*Manager, error)

func (*Manager) Cleanup 

func (m *Manager) Cleanup(sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*Manager) Close 

func (m *Manager) Close() error

func (*Manager) ConsumeClaim 

func (m *Manager) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error

ConsumeClaim push message

func (*Manager) IsAutoCommit 

func (m *Manager) IsAutoCommit() bool

func (*Manager) SendMessage 

func (m *Manager) SendMessage(topic string, key, value sarama.Encoder)

SendMessage 发送一条<key,value>到kafka指定topic中

func (*Manager) Setup 

func (m *Manager) Setup(sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type Option 

type Option func(*option)

func WithAutoSubmit 

func WithAutoSubmit(auto bool) Option

func WithClientID 

func WithClientID(CID string) Option

func WithConsumerBrokers 

func WithConsumerBrokers(brokers []string) Option

func WithConsumerGroup 

func WithConsumerGroup(group string) Option

func WithConsumerMessageBufferSize 

func WithConsumerMessageBufferSize(size int64) Option

func WithConsumerOffsetInitial 

func WithConsumerOffsetInitial(offset int64) Option

func WithConsumerOffsetsRetryMax 

func WithConsumerOffsetsRetryMax(retryMax int) Option

func WithConsumerTopics 

func WithConsumerTopics(topics []string) Option

func WithPassword 

func WithPassword(password string) Option

func WithProducerBrokers 

func WithProducerBrokers(brokers []string) Option

func WithProducerRequiredAcks 

func WithProducerRequiredAcks(acks sarama.RequiredAcks) Option

func WithProducerRetryMax 

func WithProducerRetryMax(retryMax int) Option

func WithUsername 

func WithUsername(username string) Option

func WithVersion 

func WithVersion(version sarama.KafkaVersion) Option

func WituConsumerRebalanceStrategy 

func WituConsumerRebalanceStrategy(rebalanceStrategy string) Option

type StringEncoder 

type StringEncoder string

func (StringEncoder) Encode 

func (s StringEncoder) Encode() ([]byte, error)

func (StringEncoder) Length 

func (s StringEncoder) Length() int

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.