rabbitmq

Documentation

Index

• Constants
• func MqStart() (err error)
• func ReceiveMsg(queueExchange QueueExchange, receiver Receiver, otherParams ...int) (err error)
• func Recv2(mq RabbitMQ, receiver Receiver, taskQuit chan<- struct{})
• func Send(queueExchange QueueExchange, msg interface{}) (err error)
• type MapChild
• type MapTest
• type QueueExchange
• type RabbitMQ
  • func NewMq(q QueueExchange) RabbitMQ
  • func (r *RabbitMQ) CloseMqChannel() (err error)
  • func (r *RabbitMQ) CloseMqConnect() (err error)
  • func (r *RabbitMQ) MqConnect() (err error)
  • func (r *RabbitMQ) MqOpenChannel() (err error)
  • func (mq *RabbitMQ) SendDelayMsg(body string, ttl int64) (err error)
  • func (mq *RabbitMQ) SendMsg(body interface{}) (err error)
  • func (mq *RabbitMQ) SendRetryMsg(body string, retry_nums int32, args ...string)
• type Receiver
• type TestReceive
  • func (t *TestReceive) Consumer(byte []byte) error
  • func (t *TestReceive) FailAction(err error, byte []byte) error

Constants

const (
	InitialTimeout      = 5                // 默认失败重新间隔/s
	DefaultExchangeName = "default.change" // 默认交换机
	DefaultExchangeType = "direct"         // 默认交换类型
)

Functions

func MqStart

func MqStart() (err error)

func ReceiveMsg

func ReceiveMsg(queueExchange QueueExchange, receiver Receiver, otherParams ...int) (err error)

ReceiveMsg 接收消息，连接失败后进行重试

func Recv2

func Recv2(mq RabbitMQ, receiver Receiver, taskQuit chan<- struct{})

Recv2 当前监听队列关闭后通知进行重试机制

func Send

func Send(queueExchange QueueExchange, msg interface{}) (err error)

Send 发送消息

Types

type MapChild

type MapChild struct {
	Name string `json:"name"`
	Age  int    `json:"age"`
}

type MapTest

type MapTest struct {
	Name  string   `json:"name"`
	Age   int      `json:"age"`
	Child MapChild `json:"child"`
}

type QueueExchange

type QueueExchange struct {
	QuName string // 队列名称
	RtKey  string // key值
	ExName string // 交换机名称
	ExType string // 交换机类型
	Dns    string //链接地址
}

QueueExchange 定义队列交换机对象,外部可调用

type RabbitMQ

type RabbitMQ struct {
	Channel *amqp.Channel

	QueueName    string // 队列名称
	RoutingKey   string // key名称
	ExchangeName string // 交换机名称
	ExchangeType string // 交换机类型
	// contains filtered or unexported fields
}

RabbitMQ 定义RabbitMQ对象

var StartQueue *RabbitMQ

func NewMq

func NewMq(q QueueExchange) RabbitMQ

NewMq 创建一个新的Mq操作对象，用于延迟消息处理

func (*RabbitMQ) CloseMqChannel

func (r *RabbitMQ) CloseMqChannel() (err error)

CloseMqChannel 关闭rabbitMQ信道

func (*RabbitMQ) CloseMqConnect

func (r *RabbitMQ) CloseMqConnect() (err error)

CloseMqConnect 关闭mq链接

func (*RabbitMQ) MqConnect

func (r *RabbitMQ) MqConnect() (err error)

MqConnect 获取连接给 RabbitMQ

func (*RabbitMQ) MqOpenChannel

func (r *RabbitMQ) MqOpenChannel() (err error)

MqOpenChannel 获取信道给 RabbitMQ

func (*RabbitMQ) SendDelayMsg

func (mq *RabbitMQ) SendDelayMsg(body string, ttl int64) (err error)

SendDelayMsg 生产者发送延时消息

func (*RabbitMQ) SendMsg

func (mq *RabbitMQ) SendMsg(body interface{}) (err error)

Send 生产者发送消息

func (*RabbitMQ) SendRetryMsg

func (mq *RabbitMQ) SendRetryMsg(body string, retry_nums int32, args ...string)

SendRetryMsg 发送重试消息

type Receiver

type Receiver interface {
	Consumer([]byte) error
	FailAction(error, []byte) error
}

Receiver 定义接收者接口

type TestReceive

type TestReceive struct {
}

func (*TestReceive) Consumer

func (t *TestReceive) Consumer(byte []byte) error

func (*TestReceive) FailAction

func (t *TestReceive) FailAction(err error, byte []byte) error