[Improve] pool sendRequest to improve producer perf (#1126)

### Motivation

`sendRequest ` in producer is a frequently allocated struct, pool it can decrease the memory allocation.

### Modifications

1. Init a sync.Pool;
2. Get sendRequest  from the pool when we need; 
3. Reset sendRequest and put it back into the pool when it is done.
---------

Co-authored-by: gunli <[email protected]>

Author: gunli

pulsar/producer_partition.go

@@ -60,11 +60,20 @@ var (
 	errProducerClosed     = newError(ProducerClosed, "producer already been closed")
 	errMemoryBufferIsFull = newError(ClientMemoryBufferIsFull, "client memory buffer is full")
 
-	buffersPool sync.Pool
+	buffersPool     sync.Pool
+	sendRequestPool *sync.Pool
 )
 
 var errTopicNotFount = "TopicNotFound"
 
+func init() {
+	sendRequestPool = &sync.Pool{
+		New: func() interface{} {
+			return &sendRequest{}
+		},
+	}
+}
+
 type partitionProducer struct {
 	state  uAtomic.Int32
 	client *client
@@ -527,7 +536,9 @@ func (p *partitionProducer) internalSend(sr *sendRequest) {
 		}
 		// update chunk id
 		sr.mm.ChunkId = proto.Int32(int32(chunkID))
-		nsr := &sendRequest{
+		nsr := sendRequestPool.Get().(*sendRequest)
+		*nsr = sendRequest{
+			pool:                sendRequestPool,
 			ctx:                 sr.ctx,
 			msg:                 sr.msg,
 			producer:            sr.producer,
@@ -1150,7 +1161,9 @@ func (p *partitionProducer) internalSendAsync(
 		return
 	}
 
-	sr := &sendRequest{
+	sr := sendRequestPool.Get().(*sendRequest)
+	*sr = sendRequest{
+		pool:             sendRequestPool,
 		ctx:              ctx,
 		msg:              msg,
 		producer:         p,
@@ -1395,6 +1408,7 @@ func (p *partitionProducer) Close() {
 }
 
 type sendRequest struct {
+	pool             *sync.Pool
 	ctx              context.Context
 	msg              *ProducerMessage
 	producer         *partitionProducer
@@ -1477,6 +1491,13 @@ func (sr *sendRequest) done(msgID MessageID, err error) {
 			sr.transaction.endSendOrAckOp(err)
 		}
 	}
+
+	pool := sr.pool
+	if pool != nil {
+		// reset all the fields
+		*sr = sendRequest{}
+		pool.Put(sr)
+	}
 }
 
 func (p *partitionProducer) blockIfQueueFull() bool {