lib

Documentation

Index

• Constants
• Variables
• func NewMQ()
• type Client
  • func NewClient(clientId string) *Client
  • func (client *Client) Close() error
  • func (client *Client) Connect() error
  • func (client *Client) GetClientID() string
  • func (client *Client) Publish(topic string, qos byte, retained bool, data []byte) error
  • func (client *Client) Subscribe(observer func(c *Client, msg *Message), qos byte, topics ...string) error
  • func (client *Client) Unsubscribe(topics ...string)
• type ClientManager
• type MQ
  • func GetMQ() *MQ
  • func (this *MQ) Consume(queue string, key string, callbak func(<-chan amqp.Delivery, string))
  • func (this *MQ) DecQueueAndBind(queues string, key string, exchange string) error
  • func (this *MQ) DecQueueAndBindWithArgs(queues string, key string, exchange string, args map[string]interface{}) error
  • func (this *MQ) ListenConfirm()
  • func (this *MQ) NotifyReturn()
  • func (this *MQ) SendDelayMessage(key string, exchange string, message string, delay int) error
  • func (this *MQ) SendMessage(key string, exchange string, message string) error
  • func (this *MQ) SetConfirm()
• type Message

Constants

const (
	QUEUE_NEWUSER       = "newuser"       //用户注册 对应的队列名称
	QUEUE_NEWUSER_UNION = "newuser_union" //合作单位用户注册 对应的队列名称
	EXCHANGE_USER       = "UserExchange"  //用户模块相关的交换机
	EXCHANGE_USER_DELAY = "UserExchangeDelay"
	ROUTER_KEY_USERREG  = "userreg" //注册用户的路由key

	EXCHANGE_TRANS   = "TransExchange" //转账相关交换机
	ROUTER_KEY_TRANS = "trans"         //转账相关路由key
	QUEUE_TRANS      = "TransQueueA"   //转账相关队列
)

const (
	//Host     = "192.168.31.32:1883"
	//UserName = "root"
	//Password = "root"
	Host     = "47.116.75.143:1883"
	UserName = "client"
	Password = "hi2020go"
)

mqtt服务器配置

Variables

var Manager = &ClientManager{
	clients: make(map[string]*Client),
}

Functions

func NewMQ

func NewMQ()

Types

type Client

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

func NewClient

func NewClient(clientId string) *Client

func (*Client) Close

func (client *Client) Close() error

func (*Client) Connect

func (client *Client) Connect() error

func (*Client) GetClientID

func (client *Client) GetClientID() string

func (*Client) Publish

func (client *Client) Publish(topic string, qos byte, retained bool, data []byte) error

发布消息 retained: 是否保留信息

func (*Client) Subscribe

func (client *Client) Subscribe(observer func(c *Client, msg *Message), qos byte, topics ...string) error

消费消息

func (*Client) Unsubscribe

func (client *Client) Unsubscribe(topics ...string)

type ClientManager

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

type MQ

type MQ struct {
	Channel *amqp.Channel
	// contains filtered or unexported fields
}

func GetMQ

func GetMQ() *MQ

func (*MQ) Consume

func (this *MQ) Consume(queue string, key string, callbak func(<-chan amqp.Delivery, string))

func (*MQ) DecQueueAndBind

func (this *MQ) DecQueueAndBind(queues string, key string, exchange string) error

申明队列以及绑定路由key 多个队列 可以用逗号分隔

func (*MQ) DecQueueAndBindWithArgs

func (this *MQ) DecQueueAndBindWithArgs(queues string, key string, exchange string, args map[string]interface{}) error

func (*MQ) ListenConfirm

func (this *MQ) ListenConfirm()

func (*MQ) NotifyReturn

func (this *MQ) NotifyReturn()

func (*MQ) SendDelayMessage

func (this *MQ) SendDelayMessage(key string, exchange string, message string, delay int) error

发送延迟消息

func (*MQ) SendMessage

func (this *MQ) SendMessage(key string, exchange string, message string) error

func (*MQ) SetConfirm

func (this *MQ) SetConfirm()

type Message

type Message struct {
	// client_id
	ClientID string `json:"client_id"`
	// 接口名，订阅号通过识别接口名处理相应业务
	Action string `json:"action"`
	// 数据类型
	Type string `json:"type"`
	// 发布时间
	Time int64 `json:"time"`
	// 业务数据的header，可以携带一些系统参数
	Header interface{} `json:"header"`
	// 业务数据的body，业务参数
	Body interface{} `json:"body"`
}