producer

Documentation

Index

• func NewKafkaProducer(addr string, opt Poptions) *kafkaProducer
• type AMQP
  • func NewAMQPProducer(addr string, opt Poptions) *AMQP
  • func (am *AMQP) Close()
  • func (am AMQP) DelayPublish(delay time.Duration, msg PMessage, route PRoute) error
  • func (am *AMQP) Publish(msg PMessage, route PRoute) error
• type MqProducer
  • func NewProducer(mqType mq.MQType, addr string, opt ...Option) MqProducer
• type NsqProducer
  • func NewNsqProducer(addr string, opt Poptions) *NsqProducer
  • func (nsqp NsqProducer) Close()
  • func (nsqp NsqProducer) DelayPublish(delay time.Duration, msg PMessage, route PRoute) error
  • func (nsqp NsqProducer) Publish(msg PMessage, route PRoute) error
• type Option
  • func WithProducerQueueBind(route PRoute) Option
  • func WithProducerTimeout(num time.Duration) Option
  • func WithProducerUserAndPwd(user, pwd string) Option
  • func WithProducerVhost(vhost string) Option
• type PMessage
• type PRoute
• type Poptions

Functions

func NewKafkaProducer

func NewKafkaProducer(addr string, opt Poptions) *kafkaProducer

Types

type AMQP

type AMQP struct {
	Online bool //是否在线
	// contains filtered or unexported fields
}

func NewAMQPProducer

func NewAMQPProducer(addr string, opt Poptions) *AMQP

NewAMQPProducer New 创建一个新的操作RabbitMQ的对象

func (*AMQP) Close

func (am *AMQP) Close()

func (AMQP) DelayPublish

func (am AMQP) DelayPublish(delay time.Duration, msg PMessage, route PRoute) error

func (*AMQP) Publish

func (am *AMQP) Publish(msg PMessage, route PRoute) error

type MqProducer

type MqProducer interface {
	// Publish 发送消息，有错误就返回,发送成功error为nil
	Publish(PMessage, PRoute) error
	// DelayPublish 延迟发送消息，有错误就返回,发送成功error为nil
	DelayPublish(time.Duration, PMessage, PRoute) error
	Close()
}

MqProducer 发送消息接口

func NewProducer

func NewProducer(mqType mq.MQType, addr string, opt ...Option) MqProducer

type NsqProducer

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

func NewNsqProducer

func NewNsqProducer(addr string, opt Poptions) *NsqProducer

func (NsqProducer) Close

func (nsqp NsqProducer) Close()

func (NsqProducer) DelayPublish

func (nsqp NsqProducer) DelayPublish(delay time.Duration, msg PMessage, route PRoute) error

func (NsqProducer) Publish

func (nsqp NsqProducer) Publish(msg PMessage, route PRoute) error

type Option

type Option func(opt *Poptions)

Option ProducerOption producer option

func WithProducerQueueBind

func WithProducerQueueBind(route PRoute) Option

WithProducerQueueBind 设置绑定queue信息

func WithProducerTimeout

func WithProducerTimeout(num time.Duration) Option

WithProducerTimeout 设置超时时间

func WithProducerUserAndPwd

func WithProducerUserAndPwd(user, pwd string) Option

WithProducerUserAndPwd 设置用户名密码

func WithProducerVhost

func WithProducerVhost(vhost string) Option

WithProducerVhost 设置vhost

type PMessage

type PMessage struct {
	Time time.Time //kafka 指定消息发送的时间
	Key  string    //kafka消息有key，如果是其他mq没有key则不用填写
	Body []byte    //消息内容
}

PMessage 发送的消息

type PRoute

type PRoute struct {
	Exchange  string                 //AMQP
	RouteKey  string                 //AMQP
	QueueName string                 //AMQP
	Args      map[string]interface{} //AMQP
	Topic     string                 //kafka topic, nsq topic
}

PRoute 发送消息的路由 ,不必都填写，但是要知道路由规则，每个mq都不太一样。

type Poptions

type Poptions struct {
	Vhost            string        //AMQP
	Timeout          time.Duration //秒,0不超时
	Username         string
	Password         string
	AuthSecret       string
	AMQPQueueDeclare mq.AMQPQueueDeclareOPT
	AMQPPulish       mq.AMQPPubiishOPT
	AMQPQueueBind    mq.AMQPQueueBindOPT
	GenQueueExpires  int //amq.gen--XXXXXX 自动清除时间 ms
	PRoute           PRoute
}

Poptions producer的option