Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

consumer

package

Go to main page

Versions in this module

v1

v1.0.8

Jan 31, 2024

v1.0.7

Oct 11, 2023

v1.0.7-beta

Oct 12, 2023

v1.0.6

Oct 10, 2023

Changes in this version

+ type DefaultPullConsumer struct

+ GroupName string

+ Model MessageModel

+ UnitMode bool

+ func (pc *DefaultPullConsumer) ACK(ctx context.Context, cr *ConsumeRequest, result ConsumeResult)

+ func (pc *DefaultPullConsumer) ConsumeMessageDirectly(msg *primitive.MessageExt, brokerName string) *internal.ConsumeMessageDirectlyResult

+ func (pc *DefaultPullConsumer) CurrentOffset(queue *primitive.MessageQueue) (int64, error)

+ func (pc *DefaultPullConsumer) GetClient() internal.RMQClient

+ func (pc *DefaultPullConsumer) GetConsumerRunningInfo(stack bool) *internal.ConsumerRunningInfo

+ func (pc *DefaultPullConsumer) GetConsumerStatus(topic string) *internal.ConsumerStatus

+ func (pc *DefaultPullConsumer) GetModel() string

+ func (pc *DefaultPullConsumer) GetWhere() string

+ func (pc *DefaultPullConsumer) GetcType() string

+ func (pc *DefaultPullConsumer) IsSubscribeTopicNeedUpdate(topic string) bool

+ func (pc *DefaultPullConsumer) IsUnitMode() bool

+ func (pc *DefaultPullConsumer) PersistConsumerOffset() error

+ func (pc *DefaultPullConsumer) PersistOffset(ctx context.Context, topic string) error

+ func (pc *DefaultPullConsumer) Poll(ctx context.Context, timeout time.Duration) (*ConsumeRequest, error)

+ func (pc *DefaultPullConsumer) Pull(ctx context.Context, numbers int) (*primitive.PullResult, error)

+ func (pc *DefaultPullConsumer) PullFrom(ctx context.Context, queue *primitive.MessageQueue, offset int64, numbers int) (*primitive.PullResult, error)

+ func (pc *DefaultPullConsumer) Rebalance()

+ func (pc *DefaultPullConsumer) RebalanceIfNotPaused()

+ func (pc *DefaultPullConsumer) ReleaseSubGroup(group string) bool

+ func (pc *DefaultPullConsumer) ResetOffset(topic string, table map[primitive.MessageQueue]int64)

+ func (pc *DefaultPullConsumer) ServiceRunningOk() bool

+ func (pc *DefaultPullConsumer) Shutdown() error

+ func (pc *DefaultPullConsumer) Start() error

+ func (pc *DefaultPullConsumer) Subscribe(topic string, selector MessageSelector) error

+ func (pc *DefaultPullConsumer) SubscriptionDataList() []*internal.SubscriptionData

+ func (pc *DefaultPullConsumer) Unsubscribe(topic string) error

+ func (pc *DefaultPullConsumer) UpdateOffset(queue *primitive.MessageQueue, offset int64) error

+ func (pc *DefaultPullConsumer) UpdateTopicSubscribeInfo(topic string, mqs []*primitive.MessageQueue)

v1.0.5

Oct 9, 2023

v1.0.5-beta

Oct 9, 2023

v1.0.4

Oct 7, 2023

v1.0.3

Sep 23, 2023

v1.0.2

Sep 23, 2023

v1.0.1

Sep 23, 2023

Changes in this version

+ const Mb

+ const SQL92

+ const TAG

+ var ErrNoNewMsg = errors.New("no new message found")

+ func AllocateByAveragely(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, ...) []*primitive.MessageQueue

+ func AllocateByAveragelyCircle(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, ...) []*primitive.MessageQueue

+ func AllocateByMachineNearby(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, ...) []*primitive.MessageQueue

+ func CreateReplyMessage(requestMessage *primitive.MessageExt, body []byte) (*primitive.Message, error)

+ func GetReplyToClient(msg *primitive.MessageExt) string

+ func IsNoNewMsgError(err error) bool

+ func IsTagType(exp string) bool

+ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error)

+ func NewPushConsumer(opts ...Option) (*pushConsumer, error)

+ type AllocateStrategy func(string, string, []*primitive.MessageQueue, []string) []*primitive.MessageQueue

+ func AllocateByConfig(list []*primitive.MessageQueue) AllocateStrategy

+ func AllocateByConsistentHash(virtualNodeCnt int) AllocateStrategy

+ func AllocateByMachineRoom(consumeridcs []string) AllocateStrategy

+ type ConsumeFromWhere int

+ const ConsumeFromFirstOffset

+ const ConsumeFromLastOffset

+ const ConsumeFromTimestamp

+ type ConsumeRequest struct

+ func (cr *ConsumeRequest) GetMQ() *primitive.MessageQueue

+ func (cr *ConsumeRequest) GetMsgList() []*primitive.MessageExt

+ func (cr *ConsumeRequest) GetPQ() *processQueue

+ type ConsumeResult int

+ const Commit

+ const ConsumeRetryLater

+ const ConsumeSuccess

+ const Rollback

+ const SuspendCurrentQueueAMoment

+ type ConsumeResultHolder struct

+ type ConsumeStatus struct

+ ConsumeFailedMsgs int64

+ ConsumeFailedTPS float64

+ ConsumeOKTPS float64

+ ConsumeRT float64

+ PullRT float64

+ PullTPS float64

+ type ConsumeType string

+ type ConsumerReturn int

+ const ExceptionReturn

+ const FailedReturn

+ const NullReturn

+ const SuccessReturn

+ const TimeoutReturn

+ type ExpressionType string

+ type Limiter func(topic string)

+ type MessageModel int

+ const BroadCasting

+ const Clustering

+ func (mode MessageModel) String() string

+ type MessageQueueKey primitive.MessageQueue

+ func (mq *MessageQueueKey) UnmarshalText(text []byte) error

+ func (mq MessageQueueKey) MarshalText() (text []byte, err error)

+ type MessageSelector struct

+ Expression string

+ Type ExpressionType

+ type MockOffsetStore struct

+ func NewMockOffsetStore(ctrl *gomock.Controller) *MockOffsetStore

+ func (m *MockOffsetStore) EXPECT() *MockOffsetStoreMockRecorder

+ type MockOffsetStoreMockRecorder struct

+ type OffsetSerializeWrapper struct

+ OffsetTable map[MessageQueueKey]int64

+ type OffsetStore interface

+ func NewLocalFileOffsetStore(clientID, group string) OffsetStore

+ func NewRemoteOffsetStore(group string, client internal.RMQClient, namesrv internal.Namesrvs) OffsetStore

+ type Option func(*consumerOptions)

+ func WithAutoCommit(auto bool) Option

+ func WithConsumeConcurrentlyMaxSpan(consumeConcurrentlyMaxSpan int) Option

+ func WithConsumeFromWhere(w ConsumeFromWhere) Option

+ func WithConsumeGoroutineNums(nums int) Option

+ func WithConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize int) Option

+ func WithConsumeTimeout(timeout time.Duration) Option

+ func WithConsumeTimestamp(consumeTimestamp string) Option

+ func WithConsumerModel(m MessageModel) Option

+ func WithConsumerOrder(order bool) Option

+ func WithConsumerPullTimeout(consumerPullTimeout time.Duration) Option

+ func WithCredentials(c primitive.Credentials) Option

+ func WithFilterMessageHook(hooks []hooks.FilterMessageHook) Option

+ func WithGroupName(group string) Option

+ func WithInstance(name string) Option

+ func WithInterceptor(fs ...primitive.Interceptor) Option

+ func WithLimiter(limiter Limiter) Option

+ func WithMaxReconsumeTimes(times int32) Option

+ func WithNameServer(nameServers primitive.NamesrvAddr) Option

+ func WithNameServerDomain(nameServerUrl string) Option

+ func WithNamespace(namespace string) Option

+ func WithNsResolver(resolver primitive.NsResolver) Option

+ func WithPullBatchSize(batchSize int32) Option

+ func WithPullInterval(interval time.Duration) Option

+ func WithPullThresholdForQueue(pullThresholdForQueue int64) Option

+ func WithPullThresholdForTopic(pullThresholdForTopic int) Option

+ func WithPullThresholdSizeForQueue(pullThresholdSizeForQueue int) Option

+ func WithPullThresholdSizeForTopic(pullThresholdSizeForTopic int) Option

+ func WithRebalanceLockInterval(interval time.Duration) Option

+ func WithRetry(retries int) Option

+ func WithStrategy(strategy AllocateStrategy) Option

+ func WithSuspendCurrentQueueTimeMillis(suspendT time.Duration) Option

+ func WithTrace(traceCfg *primitive.TraceConfig) Option

+ func WithUnitName(unitName string) Option

+ func WithVIPChannel(enable bool) Option

+ type PullRequest struct

+ func (pr *PullRequest) String() string

+ type PushConsumerCallback struct

+ func (callback PushConsumerCallback) UniqueID() string

+ type QueueLock struct

+ type StatsManager struct

+ func NewStatsManager() *StatsManager

+ func (mgr *StatsManager) GetConsumeStatus(group, topic string) ConsumeStatus

+ func (mgr *StatsManager) ShutDownStat()