rocketmq

Documentation

Index

• type MQMessage
  • func NewMQMessage(topicName, tagName, key string, body []byte, delayTimeLevel int) *MQMessage
• type MqCustomer
  • func NewPushCustomer(groupName string, NameSrvAddr []string, isBroadCasting bool) *MqCustomer
  • func (mq *MqCustomer) ShutdownPushConsumer() error
  • func (mq *MqCustomer) Subscribe(topicName, tag string)
• type MqProducer
  • func NewMQProducer(topicName, groupName string, NameSrvAddr []string) *MqProducer
  • func (mq *MqProducer) SendAsyncMessage(msg *MQMessage, ...) (bool, error)
  • func (mq *MqProducer) SendAsyncMessageBody(msg *MQMessage) (bool, error)
  • func (mq *MqProducer) SendSyncMessage(msg *MQMessage) (bool, error)
  • func (mq *MqProducer) SendSyncbatchMessage(msgs []*MQMessage) (bool, error)
  • func (mq *MqProducer) Shutdown() error
• type RockerMQAdmin
  • func NewMQAdmin(Topic, BrokerAddr string, nameSrvAddr []string) *RockerMQAdmin
  • func (r *RockerMQAdmin) CloseMQAdmin() error
  • func (r *RockerMQAdmin) CreateTopic(TopicName string) error
  • func (r *RockerMQAdmin) DeleteTopic(TopicName string) error
• type SendMsg
  • func ConnRockerMQ(ip, port, topic, group string) *SendMsg
  • func (p *SendMsg) Close()
  • func (p *SendMsg) DelaySendSync(dataMsg []byte)
  • func (p *SendMsg) SendAsync(dataMsg []byte)
  • func (p *SendMsg) SendAsyncBatch(msgs []*primitive.Message)
  • func (p *SendMsg) SendSync(dataMsg []byte)
  • func (p *SendMsg) TagSendSync(tag string, dataMsg []byte)

Types

type MQMessage

type MQMessage struct {
	TopicName      string
	Body           []byte
	TagName        string
	Key            string
	DelayTimeLevel int
}

func NewMQMessage

func NewMQMessage(topicName, tagName, key string, body []byte, delayTimeLevel int) *MQMessage

type MqCustomer

type MqCustomer struct {
	QueuePull   primitive.MessageQueue
	MessageList chan *primitive.MessageExt
	// contains filtered or unexported fields
}

func NewPushCustomer

func NewPushCustomer(groupName string, NameSrvAddr []string, isBroadCasting bool) *MqCustomer

NewPushCustomer 创建一个MQ消费者

func (*MqCustomer) ShutdownPushConsumer

func (mq *MqCustomer) ShutdownPushConsumer() error

ShutdownPushConsumer 关闭消费者

func (*MqCustomer) Subscribe

func (mq *MqCustomer) Subscribe(topicName, tag string)

Subscribe 订阅者模式

type MqProducer

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

func NewMQProducer

func NewMQProducer(topicName, groupName string, NameSrvAddr []string) *MqProducer

NewMQProducer 生产者

func (*MqProducer) SendAsyncMessage

func (mq *MqProducer) SendAsyncMessage(msg *MQMessage, callback func(ctx context.Context, result *primitive.SendResult, err error)) (bool, error)

SendAsyncMessage 发送异步消息后回调

func (*MqProducer) SendAsyncMessageBody

func (mq *MqProducer) SendAsyncMessageBody(msg *MQMessage) (bool, error)

SendAsyncMessageBody 发送异步消息后回调

func (*MqProducer) SendSyncMessage

func (mq *MqProducer) SendSyncMessage(msg *MQMessage) (bool, error)

SendSyncMessage 发送同步消息

func (*MqProducer) SendSyncbatchMessage

func (mq *MqProducer) SendSyncbatchMessage(msgs []*MQMessage) (bool, error)

SendSyncbatchMessage SendSyncMessageList 发送批量消息

func (*MqProducer) Shutdown

func (mq *MqProducer) Shutdown() error

Shutdown 关闭

type RockerMQAdmin

type RockerMQAdmin struct {
	NameSrvAddr []string
	BrokerAddr  string
	Admin       admin.Admin
	Topic       string
	// contains filtered or unexported fields
}

func NewMQAdmin

func NewMQAdmin(Topic, BrokerAddr string, nameSrvAddr []string) *RockerMQAdmin

NewMQAdmin 创建MQAdmin

func (*RockerMQAdmin) CloseMQAdmin

func (r *RockerMQAdmin) CloseMQAdmin() error

CloseMQAdmin 关闭MQAdmin

func (*RockerMQAdmin) CreateTopic

func (r *RockerMQAdmin) CreateTopic(TopicName string) error

CreateTopic 创建topic

func (*RockerMQAdmin) DeleteTopic

func (r *RockerMQAdmin) DeleteTopic(TopicName string) error

DeleteTopic 删除topic

type SendMsg

type SendMsg struct {
	IP       string
	Port     string
	Topic    string
	Group    string
	Producer rocketmq.Producer
}

func ConnRockerMQ

func ConnRockerMQ(ip, port, topic, group string) *SendMsg

func (*SendMsg) Close

func (p *SendMsg) Close()

Close 关闭

func (*SendMsg) DelaySendSync

func (p *SendMsg) DelaySendSync(dataMsg []byte)

DelaySendSync // 延迟发送消息 delay send message

func (*SendMsg) SendAsync

func (p *SendMsg) SendAsync(dataMsg []byte)

SendAsync 发送消息后回调 TODO 有高并发bug

func (*SendMsg) SendAsyncBatch

func (p *SendMsg) SendAsyncBatch(msgs []*primitive.Message)

SendAsyncBatch // 批量发送消息

func (*SendMsg) SendSync

func (p *SendMsg) SendSync(dataMsg []byte)

SendSync 发送异步消息

func (*SendMsg) TagSendSync

func (p *SendMsg) TagSendSync(tag string, dataMsg []byte)

TagSendSync 发送带有tag的消息