[ISSUE #1112] feat: optimize producer send async (#1111)

* feat: optimize producer send async

* fix: fix mq override bug

Author: twz915

internal/remote/remote_client.go

@@ -112,23 +112,24 @@ func (c *remotingClient) InvokeSync(ctx context.Context, addr string, request *R
 
 // InvokeAsync send request without blocking, just return immediately.
 func (c *remotingClient) InvokeAsync(ctx context.Context, addr string, request *RemotingCommand, callback func(*ResponseFuture)) error {
-	conn, err := c.connect(ctx, addr)
-	if err != nil {
-		return err
-	}
-
 	resp := NewResponseFuture(ctx, request.Opaque, callback)
 	c.responseTable.Store(resp.Opaque, resp)
 
-	err = c.sendRequest(ctx, conn, request)
-	if err != nil {
-		c.responseTable.Delete(request.Opaque)
-		return err
-	}
-
 	go primitive.WithRecover(func() {
+		defer resp.executeInvokeCallback()
+		defer c.responseTable.Delete(request.Opaque)
+
+		conn, err := c.connect(ctx, addr)
+		if err != nil {
+			resp.Err = err
+			return
+		}
+		err = c.sendRequest(ctx, conn, request)
+		if err != nil {
+			resp.Err = err
+			return
+		}
 		c.receiveAsync(resp)
-		c.responseTable.Delete(request.Opaque)
 	})
 
 	return nil

producer/producer.go

@@ -329,7 +329,7 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 		if mq != nil {
 			lastBrokerName = mq.BrokerName
 		}
-		mq := p.selectMessageQueue(msg, lastBrokerName)
+		mq = p.selectMessageQueue(msg, lastBrokerName)
 		if mq == nil {
 			err = fmt.Errorf("the topic=%s route info not found", msg.Topic)
 			continue