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 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, err error)
• type MqcResover
• type MqpResover
• type MsgItem
  • func (w *MsgItem) Body() map[string]interface{}
  • func (w *MsgItem) Header() map[string]string
  • func (w *MsgItem) String() string
• type MsgOption
  • func WithHeader(key, val string) MsgOption
  • func WithXRequestID(reqId string) MsgOption
• type MsgWrap
  • func (w *MsgWrap) Body() map[string]interface{}
  • func (w *MsgWrap) Header() map[string]string
  • func (w *MsgWrap) String() string
• type Option
  • func WithDBIndex(idx int) Option
  • func WithOption(key string, val any) Option
  • func WithPoolSize(size int) Option
• type 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 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()
	Close()
}

IMQC consumer接口

func NewMQC

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

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

type IMQCMessage

type IMQCMessage interface {
	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
	Count(key string) (int64, error)
}

IQueue 消息队列

type Message

type Message interface {
	Header() map[string]string
	Body() map[string]interface{}
	String() string
}

func NewMsg

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

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"`
	BodyMap   xtypes.XMap `json:"body"`
	// contains filtered or unexported fields
}

func (*MsgItem) Body

func (w *MsgItem) Body() map[string]interface{}

func (*MsgItem) Header

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

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"`
	BodyMap   xtypes.XMap `json:"body"`
	// contains filtered or unexported fields
}

func (*MsgWrap) Body

func (w *MsgWrap) Body() map[string]interface{}

func (*MsgWrap) Header

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

func (*MsgWrap) String

func (w *MsgWrap) String() string

type Option

type Option func(*Options)

Option 配置选项

func WithDBIndex

func WithDBIndex(idx int) Option

func WithOption

func WithOption(key string, val any) Option

func WithPoolSize

func WithPoolSize(size int) Option

type Options

type Options struct {
	CfgData map[string]any
}

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
}