queue

Documentation

Index

• Constants
• Variables
• func DeregisterProducer(name string)
• func DeregisterrConsumer(name string)
• func NewBuilder() container.StandardBuilder
• func RegisterConsumer(resolver MqcResover)
• func RegisterProducer(resolver MqpResover)
• type ConsumeCallback
• type DelayCallback
• type DelayProcessor
• type IComponentQueue
• type IMQC
  • func NewMQC(proto, configName string, setting config.Config) (IMQC, error)
• type IMQCMessage
• type IMQP
  • func NewMQP(proto string, setting config.Config, opts ...Option) (IMQP, error)
• type IQueue
• type Message
  • func NewMsg(obj interface{}, opts ...MsgOption) (msg Message)
• type MqcResover
• type MqpResover
• type MsgItem
  • func (w MsgItem) Body() []byte
  • func (w MsgItem) Header() map[string]string
  • func (w MsgItem) MarshalBinary() (data []byte, err error)
  • func (w *MsgItem) String() string
• type MsgOption
  • func WithHeader(key, val string) MsgOption
  • func WithXRequestID(reqId string) MsgOption
• type MsgWrap
  • func (w *MsgWrap) Body() []byte
  • func (w *MsgWrap) Header() map[string]string
  • func (w MsgWrap) MarshalBinary() (data []byte, err error)
  • func (w MsgWrap) String() string
• type Option
  • func WithOption(key string, val any) Option
• type Options
  • func DefaultOptions() *Options
• type StandardQueue
  • func NewStandardQueue(c container.Container) StandardQueue
• type TaskInfo

Constants

const (
	QueueKey string = "qk"
)

const (
	TypeNode = "queues"
)

Variables

var DefaultMaxQueueLen = 100

默认最大队列长度 100

var (
	MaxRetrtCount int64 = 3 //最大重试三次
)

var Nil error = errors.New("queue nil")

Functions

func DeregisterProducer

func DeregisterProducer(name string)

Deregister 清理配置适配器

func DeregisterrConsumer

func DeregisterrConsumer(name string)

Deregister 清理配置适配器

func NewBuilder

func NewBuilder() container.StandardBuilder

func RegisterConsumer

func RegisterConsumer(resolver MqcResover)

RegisterConsumer 注册消息消费

func RegisterProducer

func RegisterProducer(resolver MqpResover)

RegisterProducer 注册配置文件适配器

Types

type ConsumeCallback

type ConsumeCallback func(IMQCMessage)

type DelayCallback

type DelayCallback func(key string, msgList ...Message) error

DelayCallback 延迟消息处理回调

type DelayProcessor

type DelayProcessor interface {
	Start(done chan struct{}) error
	AppendMessage(msg Message, delaySeconds int64) error
}

DelayProcessor 延迟消息队列处理器

type IComponentQueue

type IComponentQueue interface {
	GetQueue(name string) (q IQueue)
}

IComponentQueue Component Queue

type IMQC

type IMQC interface {
	Connect() error
	Consume(task TaskInfo, callback ConsumeCallback) (err error)
	Unconsume(queue string)
	Start() error
	Close() error
}

IMQC consumer接口

func NewMQC

func NewMQC(proto, configName string, setting config.Config) (IMQC, error)

NewMQC 根据适配器名称及参数返回配置处理器

type IMQCMessage

type IMQCMessage interface {
	MessageId() string
	RetryCount() int64
	Ack() error
	Nack(error) error
	Original() string
	GetMessage() Message
}

IMQCMessage 队列消息

type IMQP

type IMQP interface {
	Push(key string, value Message) error
	DelayPush(key string, value Message, delaySeconds int64) error
	//Count(key string) (int64, error)
	Close() error
}

IMQP 消息生产

func NewMQP

func NewMQP(proto string, setting config.Config, opts ...Option) (IMQP, error)

NewMQP 根据适配器名称及参数返回配置处理器

type IQueue

type IQueue interface {
	Send(ctx context.Context, key string, value interface{}) error
	DelaySend(ctx context.Context, key string, value interface{}, delaySeconds int64) error
}

IQueue 消息队列

type Message

type Message interface {
	encoding.BinaryMarshaler
	Header() map[string]string
	Body() []byte
	String() string
}

func NewMsg

func NewMsg(obj interface{}, opts ...MsgOption) (msg Message)

type MqcResover

type MqcResover interface {
	Name() string
	Resolve(configName string, setting config.Config) (IMQC, error)
}

mqcResover 定义消息消费解析器

type MqpResover

type MqpResover interface {
	Name() string
	Resolve(setting config.Config, opts ...Option) (IMQP, error)
}

MqpResover 定义配置文件转换方法

type MsgItem

type MsgItem struct {
	HeaderMap xtypes.SMap     `json:"header"`
	BodyBytes json.RawMessage `json:"body"`
	ItemBytes json.RawMessage `json:"-"`
}

func (MsgItem) Body

func (w MsgItem) Body() []byte

func (MsgItem) Header

func (w MsgItem) Header() map[string]string

func (MsgItem) MarshalBinary

func (w MsgItem) MarshalBinary() (data []byte, err error)

func (*MsgItem) String

func (w *MsgItem) String() string

type MsgOption

type MsgOption func(m *MsgWrap)

func WithHeader

func WithHeader(key, val string) MsgOption

func WithXRequestID

func WithXRequestID(reqId string) MsgOption

type MsgWrap

type MsgWrap struct {
	HeaderMap xtypes.SMap     `json:"header,omitempty"`
	BodyBytes json.RawMessage `json:"body"`
	// contains filtered or unexported fields
}

func (*MsgWrap) Body

func (w *MsgWrap) Body() []byte

func (*MsgWrap) Header

func (w *MsgWrap) Header() map[string]string

func (MsgWrap) MarshalBinary

func (w MsgWrap) MarshalBinary() (data []byte, err error)

func (MsgWrap) String

func (w MsgWrap) String() string

type Option

type Option func(*Options)

Option 配置选项

func WithOption

func WithOption(key string, val any) Option

type Options

type Options struct {
	CfgData map[string]any
}

func DefaultOptions

func DefaultOptions() *Options

type StandardQueue

type StandardQueue interface {
	GetQueue(name string, opts ...Option) (q IQueue)
}

func NewStandardQueue

func NewStandardQueue(c container.Container) StandardQueue

NewStandardQueue 创建queue

type TaskInfo

type TaskInfo interface {
	GetQueue() string
	GetConcurrency() int
	GetVisibilityTimeout() int
	GetBufferSize() int
	GetMeta() metadata.Metadata
}