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.2.7

Jan 15, 2020

v1.2.6

Jan 14, 2020

v1.2.5

Jan 13, 2020

Changes in this version

+ const Mb

+ const SQL92

+ const TAG

+ var ErrBrokerNotFound = errors.New("broker can not found")

+ var ErrCreated = errors.New("consumer group has been created")

+ 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 IsTagType(exp string) bool

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

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

+ func ShutDownStatis()

+ 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 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

+ func GetConsumeStatus(group, topic string) ConsumeStatus

+ type ConsumeType string

+ type ConsumerReturn int

+ const ExceptionReturn

+ const FailedReturn

+ const NullReturn

+ const SuccessReturn

+ const TimeoutReturn

+ type ExpressionType 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 WithConsumeFromWhere(w ConsumeFromWhere) Option

+ func WithConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize int) Option

+ func WithConsumerModel(m MessageModel) Option

+ func WithConsumerOrder(order bool) Option

+ func WithCredentials(c primitive.Credentials) Option

+ func WithGroupName(group string) Option

+ func WithInstance(name string) Option

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

+ func WithMaxReconsumeTimes(times int32) Option

+ func WithNameServer(nameServers primitive.NamesrvAddr) Option

+ func WithNameServerDomain(nameServerUrl string) Option

+ func WithNamespace(namespace string) Option

+ func WithPullBatchSize(batchSize int32) Option

+ func WithPullInterval(interval time.Duration) 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 WithVIPChannel(enable bool) Option

+ type PullConsumer interface

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

+ PersistOffset func(ctx context.Context) error

+ Pull func(ctx context.Context, topic string, selector MessageSelector, numbers int) (*primitive.PullResult, error)

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

+ Shutdown func()

+ Start func()

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

+ type PullRequest struct

+ func (pr *PullRequest) String() string

+ type PushConsumerCallback struct

+ func (callback PushConsumerCallback) UniqueID() string

+ type QueueLock struct

Other modules containing this package

github.com/swift9/rocketmq-client-go/v2