搜云库技术团队
序
本文主要研究一下rocketmq-client-go的defaultProducer
defaultProducer
rocketmq-client-go-v2.0.0/producer/producer.go
type defaultProducer struct {
group string
client internal.RMQClient
state int32
options producerOptions
publishInfo sync.Map
callbackCh chan interface{}
interceptor primitive.Interceptor
}
- defaultProducer定义了group、client、state、options、publishInfo、callbackCh、interceptor
NewDefaultProducer
rocketmq-client-go-v2.0.0/producer/producer.go
func NewDefaultProducer(opts ...Option) (*defaultProducer, error) {
defaultOpts := defaultProducerOptions()
for _, apply := range opts {
apply(&defaultOpts)
}
srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
if err != nil {
return nil, errors.Wrap(err, "new Namesrv failed.")
}
if !defaultOpts.Credentials.IsEmpty() {
srvs.SetCredentials(defaultOpts.Credentials)
}
defaultOpts.Namesrv = srvs
producer := &defaultProducer{
group: defaultOpts.GroupName,
callbackCh: make(chan interface{}),
options: defaultOpts,
}
producer.client = internal.GetOrNewRocketMQClient(defaultOpts.ClientOptions, producer.callbackCh)
producer.interceptor = primitive.ChainInterceptors(producer.options.Interceptors...)
return producer, nil
}
- NewDefaultProducer方法通过internal.NewNamesrv创建NameServerAddrs，之后实例化defaultProducer，然后实例化internal.GetOrNewRocketMQClient及primitive.ChainInterceptors
Start
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) Start() error {
atomic.StoreInt32(&p.state, int32(internal.StateRunning))
if len(p.options.NameServerAddrs) == 0 {
p.options.Namesrv.UpdateNameServerAddress(p.options.NameServerDomain, p.options.InstanceName)
}
p.client.RegisterProducer(p.group, p)
p.client.Start()
return nil
}
- Start方法之执行p.client.RegisterProducer及p.client.Start()
Shutdown
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) Shutdown() error {
atomic.StoreInt32(&p.state, int32(internal.StateShutdown))
p.client.UnregisterProducer(p.group)
p.client.Shutdown()
return nil
}
- Shutdown方法执行p.client.UnregisterProducer及p.client.Shutdown()
SendSync
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) SendSync(ctx context.Context, msgs ...*primitive.Message) (*primitive.SendResult, error) {
if err := p.checkMsg(msgs...); err != nil {
return nil, err
}
msg := p.encodeBatch(msgs...)
resp := new(primitive.SendResult)
if p.interceptor != nil {
primitive.WithMethod(ctx, primitive.SendSync)
producerCtx := &primitive.ProducerCtx{
ProducerGroup: p.group,
CommunicationMode: primitive.SendSync,
BornHost: utils.LocalIP,
Message: *msg,
SendResult: resp,
}
ctx = primitive.WithProducerCtx(ctx, producerCtx)
err := p.interceptor(ctx, msg, resp, func(ctx context.Context, req, reply interface{}) error {
var err error
realReq := req.(*primitive.Message)
realReply := reply.(*primitive.SendResult)
err = p.sendSync(ctx, realReq, realReply)
return err
})
return resp, err
}
err := p.sendSync(ctx, msg, resp)
return resp, err
}
- SendSync方法首先通过p.checkMsg校验消息，然后通过p.encodeBatch编码，之后对于p.interceptor不为null的执行p.interceptor，最后执行p.sendSync(ctx, msg, resp)
sendSync
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message, resp *primitive.SendResult) error {
retryTime := 1 + p.options.RetryTimes
var (
err error
)
if p.options.Namespace != "" {
msg.Topic = p.options.Namespace + "%" + msg.Topic
}
var producerCtx *primitive.ProducerCtx
for retryCount := 0; retryCount < retryTime; retryCount++ {
mq := p.selectMessageQueue(msg)
if mq == nil {
err = fmt.Errorf("the topic=%s route info not found", msg.Topic)
continue
}
addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
if addr == "" {
return fmt.Errorf("topic=%s route info not found", mq.Topic)
}
if p.interceptor != nil {
producerCtx = primitive.GetProducerCtx(ctx)
producerCtx.BrokerAddr = addr
producerCtx.MQ = *mq
}
res, _err := p.client.InvokeSync(ctx, addr, p.buildSendRequest(mq, msg), 3*time.Second)
if _err != nil {
err = _err
continue
}
return p.client.ProcessSendResponse(mq.BrokerName, res, resp, msg)
}
return err
}
- sendSync会重试retryCount，每次是先通过p.selectMessageQueue(msg)选择mq，然后通过p.options.Namesrv.FindBrokerAddrByName寻找addr，最后执行p.client.InvokeSync(ctx, addr, p.buildSendRequest
SendAsync
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) SendAsync(ctx context.Context, f func(context.Context, *primitive.SendResult, error), msgs ...*primitive.Message) error {
if err := p.checkMsg(msgs...); err != nil {
return err
}
msg := p.encodeBatch(msgs...)
if p.interceptor != nil {
primitive.WithMethod(ctx, primitive.SendAsync)
return p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error {
return p.sendAsync(ctx, msg, f)
})
}
return p.sendAsync(ctx, msg, f)
}
- SendAsync方法主要是执行p.sendAsync(ctx, msg, f)
sendAsync
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) sendAsync(ctx context.Context, msg *primitive.Message, h func(context.Context, *primitive.SendResult, error)) error {
if p.options.Namespace != "" {
msg.Topic = p.options.Namespace + "%" + msg.Topic
}
mq := p.selectMessageQueue(msg)
if mq == nil {
return errors.Errorf("the topic=%s route info not found", msg.Topic)
}
addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
if addr == "" {
return errors.Errorf("topic=%s route info not found", mq.Topic)
}
ctx, _ = context.WithTimeout(ctx, 3*time.Second)
return p.client.InvokeAsync(ctx, addr, p.buildSendRequest(mq, msg), func(command *remote.RemotingCommand, err error) {
resp := new(primitive.SendResult)
if err != nil {
h(ctx, nil, err)
} else {
p.client.ProcessSendResponse(mq.BrokerName, command, resp, msg)
h(ctx, resp, nil)
}
})
}
- sendAsync主要是执行p.client.InvokeAsync
SendOneWay
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) SendOneWay(ctx context.Context, msgs ...*primitive.Message) error {
if err := p.checkMsg(msgs...); err != nil {
return err
}
msg := p.encodeBatch(msgs...)
if p.interceptor != nil {
primitive.WithMethod(ctx, primitive.SendOneway)
return p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error {
return p.SendOneWay(ctx, msg)
})
}
return p.sendOneWay(ctx, msg)
}
- SendOneWay主要是执行p.sendOneWay(ctx, msg)
sendOneWay
rocketmq-client-go-v2.0.0/producer/producer.go
func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message) error {
retryTime := 1 + p.options.RetryTimes
if p.options.Namespace != "" {
msg.Topic = p.options.Namespace + "%" + msg.Topic
}
var err error
for retryCount := 0; retryCount < retryTime; retryCount++ {
mq := p.selectMessageQueue(msg)
if mq == nil {
err = fmt.Errorf("the topic=%s route info not found", msg.Topic)
continue
}
addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
if addr == "" {
return fmt.Errorf("topic=%s route info not found", mq.Topic)
}
_err := p.client.InvokeOneWay(ctx, addr, p.buildSendRequest(mq, msg), 3*time.Second)
if _err != nil {
err = _err
continue
}
return nil
}
return err
}
- sendOneWay主要是重试执行p.client.InvokeOneWay
小结
defaultProducer定义了group、client、state、options、publishInfo、callbackCh、interceptor；它提供了NewDefaultProducer、Start、Shutdown、SendSync、SendAsync、SendOneWay方法
doc
- producer
