queue

Documentation

Index

• type Memory
  • func NewMemory(poolNum uint) *Memory
  • func (m *Memory) Consumer(ctx context.Context, name string, f storage.ConsumerFunc, ...)
  • func (m *Memory) Publish(ctx context.Context, message storage.Messager, ...) error
  • func (m *Memory) Run(ctx context.Context)
  • func (m *Memory) Shutdown(ctx context.Context)
  • func (*Memory) String() string
• type Message
  • func (m *Message) GetErrorCount() uint64
  • func (m *Message) GetId() string
  • func (m *Message) GetPrefix() (prefix string)
  • func (m *Message) GetRoutingKey() string
  • func (m *Message) GetValues() map[string]interface{}
  • func (m *Message) SetErrorCount(count uint64)
  • func (m *Message) SetErrorIncr()
  • func (m *Message) SetId(id string)
  • func (m *Message) SetPrefix(prefix string)
  • func (m *Message) SetRoutingKey(routingKey string)
  • func (m *Message) SetValues(values map[string]interface{})
• type NSQ
  • func NewNSQ(addresses []string, cfg *nsq.Config, channelPrefix string) (*NSQ, error)
  • func (e *NSQ) Consumer(ctx context.Context, name string, f storage.ConsumerFunc, ...)
  • func (e *NSQ) Publish(ctx context.Context, message storage.Messager, ...) error
  • func (e *NSQ) Run(ctx context.Context)
  • func (e *NSQ) Shutdown(ctx context.Context)
  • func (NSQ) String() string
• type RabbitMQ
  • func NewRabbitMQ(ctx context.Context, dsn string, cfg *rabbitmq.Config) (*RabbitMQ, error)
  • func (r *RabbitMQ) Consumer(ctx context.Context, queueName string, f storage.ConsumerFunc, ...)
  • func (r *RabbitMQ) Publish(ctx context.Context, message storage.Messager, ...) error
  • func (r *RabbitMQ) Run(ctx context.Context)
  • func (r *RabbitMQ) Shutdown(ctx context.Context)
  • func (RabbitMQ) String() string
• type Redis
  • func NewRedis(producerOptions *redisqueue.ProducerOptions, ...) (*Redis, error)
  • func (r *Redis) Consumer(ctx context.Context, name string, f storage.ConsumerFunc, ...)
  • func (r *Redis) Publish(ctx context.Context, message storage.Messager, ...) error
  • func (r *Redis) Run(ctx context.Context)
  • func (r *Redis) Shutdown(ctx context.Context)
  • func (Redis) String() string
• type RocketConsumerOptions
• type RocketMQ
  • func NewRocketMQ(ctx context.Context, urls []string, consumerOptions *RocketConsumerOptions, ...) (*RocketMQ, error)
  • func (r *RocketMQ) Consumer(ctx context.Context, topicName string, f storage.ConsumerFunc, ...)
  • func (r *RocketMQ) Publish(ctx context.Context, message storage.Messager, ...) error
  • func (r *RocketMQ) Run(ctx context.Context)
  • func (r *RocketMQ) Shutdown(ctx context.Context)
  • func (RocketMQ) String() string
• type RocketProducerOptions

Types

type Memory

type Memory struct {
	PoolNum uint
	// contains filtered or unexported fields
}

func NewMemory

func NewMemory(poolNum uint) *Memory

NewMemory 内存模式

func (*Memory) Consumer

func (m *Memory) Consumer(ctx context.Context, name string, f storage.ConsumerFunc, optionFuncs ...func(*storage.ConsumeOptions))

Consumer 监听消费者

func (*Memory) Publish

func (m *Memory) Publish(ctx context.Context, message storage.Messager, optionFuncs ...func(*storage.PublishOptions)) error

Publish 消息入生产者

func (*Memory) Run

func (m *Memory) Run(ctx context.Context)

func (*Memory) Shutdown

func (m *Memory) Shutdown(ctx context.Context)

func (*Memory) String

func (*Memory) String() string

type Message

type Message struct {
	Id         string
	RoutingKey string
	Values     map[string]interface{}
	GroupId    string
	ErrorCount uint64
}

func (*Message) GetErrorCount

func (m *Message) GetErrorCount() uint64

func (*Message) GetId

func (m *Message) GetId() string

func (*Message) GetPrefix

func (m *Message) GetPrefix() (prefix string)

func (*Message) GetRoutingKey

func (m *Message) GetRoutingKey() string

func (*Message) GetValues

func (m *Message) GetValues() map[string]interface{}

func (*Message) SetErrorCount

func (m *Message) SetErrorCount(count uint64)

func (*Message) SetErrorIncr

func (m *Message) SetErrorIncr()

func (*Message) SetId

func (m *Message) SetId(id string)

func (*Message) SetPrefix

func (m *Message) SetPrefix(prefix string)

func (*Message) SetRoutingKey

func (m *Message) SetRoutingKey(routingKey string)

func (*Message) SetValues

func (m *Message) SetValues(values map[string]interface{})

type NSQ

type NSQ struct {
	// contains filtered or unexported fields
}

func NewNSQ

func NewNSQ(addresses []string, cfg *nsq.Config, channelPrefix string) (*NSQ, error)

NewNSQ nsq模式 只能监听一个channel

func (*NSQ) Consumer

func (e *NSQ) Consumer(ctx context.Context, name string, f storage.ConsumerFunc, optionFuncs ...func(*storage.ConsumeOptions))

Consumer 监听消费者

func (*NSQ) Publish

func (e *NSQ) Publish(ctx context.Context, message storage.Messager, optionFuncs ...func(*storage.PublishOptions)) error

Publish 消息入生产者

func (*NSQ) Run

func (e *NSQ) Run(ctx context.Context)

func (*NSQ) Shutdown

func (e *NSQ) Shutdown(ctx context.Context)

func (NSQ) String

func (NSQ) String() string

String 字符串类型

type RabbitMQ

type RabbitMQ struct {
	Url     string
	Handler []rabbitmq.Handler
	Config  rabbitmq.Config

	ConsumerOptions *rabbitmq.ConsumerOptions

	PublisherOptions *rabbitmq.PublisherOptions
	// contains filtered or unexported fields
}

RabbitMQ cache implement

func NewRabbitMQ

func NewRabbitMQ(
	ctx context.Context,
	dsn string,
	cfg *rabbitmq.Config,
) (*RabbitMQ, error)

func (*RabbitMQ) Consumer

func (r *RabbitMQ) Consumer(ctx context.Context, queueName string, f storage.ConsumerFunc, optionFuncs ...func(*storage.ConsumeOptions))

Consumer 监听消费者

func (*RabbitMQ) Publish

func (r *RabbitMQ) Publish(ctx context.Context, message storage.Messager, optionFuncs ...func(*storage.PublishOptions)) error

Publish 消息入生产者

func (*RabbitMQ) Run

func (r *RabbitMQ) Run(ctx context.Context)

func (*RabbitMQ) Shutdown

func (r *RabbitMQ) Shutdown(ctx context.Context)

func (RabbitMQ) String

func (RabbitMQ) String() string

type Redis

type Redis struct {
	// contains filtered or unexported fields
}

Redis cache implement

func NewRedis

func NewRedis(producerOptions *redisqueue.ProducerOptions,
	consumerOptions *redisqueue.ConsumerOptions) (*Redis, error)

NewRedis redis模式

func (*Redis) Consumer

func (r *Redis) Consumer(ctx context.Context, name string, f storage.ConsumerFunc, optionFuncs ...func(*storage.ConsumeOptions))

Consumer 监听消费者

func (*Redis) Publish

func (r *Redis) Publish(ctx context.Context, message storage.Messager, optionFuncs ...func(*storage.PublishOptions)) error

Publish 消息入生产者

func (*Redis) Run

func (r *Redis) Run(ctx context.Context)

func (*Redis) Shutdown

func (r *Redis) Shutdown(ctx context.Context)

func (Redis) String

func (Redis) String() string

type RocketConsumerOptions

type RocketConsumerOptions struct {
	GroupName         string
	MaxReconsumeTimes int32
}

type RocketMQ

type RocketMQ struct {
	Urls []string

	ConsumerOptions *RocketConsumerOptions

	ProducerOptions *RocketProducerOptions
	Credentials     primitive.Credentials
	// contains filtered or unexported fields
}

RocketMQ cache implement

func NewRocketMQ

func NewRocketMQ(
	ctx context.Context,
	urls []string,
	consumerOptions *RocketConsumerOptions,
	producerOptions *RocketProducerOptions,
	credentials *primitive.Credentials,
) (*RocketMQ, error)

func (*RocketMQ) Consumer

func (r *RocketMQ) Consumer(ctx context.Context, topicName string, f storage.ConsumerFunc, optionFuncs ...func(*storage.ConsumeOptions))

Consumer 监听消费者

func (*RocketMQ) Publish

func (r *RocketMQ) Publish(ctx context.Context, message storage.Messager, optionFuncs ...func(*storage.PublishOptions)) error

Publish 消息入生产者

func (*RocketMQ) Run

func (r *RocketMQ) Run(ctx context.Context)

func (*RocketMQ) Shutdown

func (r *RocketMQ) Shutdown(ctx context.Context)

func (RocketMQ) String

func (RocketMQ) String() string

type RocketProducerOptions

type RocketProducerOptions struct {
	GroupName  string
	RetryTimes int
}