When use pulsar go client to consume message, found that the message consumption rate is fluctuating?

#### Expected behavior / Actual behavior
I use pulsar-client-go write a demo program to consume message from pulsar, found that the **message consumption rate is fluctuating** and consume performance is very low.
But I think the consumption rate should be stable.

When use python pulsar api to make a demo to consume the same topic, the message consumption rate is stable, and performance is better than the go demo, confused...

In addition, I found that the message consumption rate of go demo will keep falling at first , after a few minutes, the rate will increase. and then keep looping this process again. When the consumption rate is low, the the CPU occupancy rate of the go demo process is also low, only around 20%.
At the same time, the python demo's CPU occupancy rate is always around 120%.

**Maybe I used the pulsar-client-go api incorrectly**? I also try use more go goroutine to receive or create more shared consumer consume together. But it doesn't seem to improve a lot.

---

#### Steps to reproduce
I have produced some messages of the same size (the size is about 2k) in advance, the data is randomly generated, and will not produce during the consumption of the demo program. Then start the demo programe

here is my code

Go demo
```go
package main

import (
"context"
"github.com/apache/pulsar-client-go/pulsar"
"log"
"strconv"
"sync/atomic"
"time"
)

var count int32

func ShowConsumeNum() {
ticker := time.NewTicker(time.Duration(60) * time.Second)
defer ticker.Stop()

for range ticker.C {
log.Println("consume " + strconv.Itoa(int(atomic.LoadInt32(&count))) + " msg")
atomic.StoreInt32(&count, 0)
}
}

func main() {
go ShowConsumeNum()

client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "my url"})
if err != nil {
log.Fatal(err)
}

defer client.Close()

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
Topic: "my_topic",
SubscriptionName: "test_consumer_" + strconv.FormatInt(time.Now().Unix(), 10),
Type: pulsar.Shared,
SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
})
if err != nil {
log.Fatal(err)
}
defer consumer.Close()

atomic.StoreInt32(&count, 0)

for {
msg, err := consumer.Receive(context.Background())
if err != nil {
log.Fatal(err)
}
//fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
msg.ID(), string(msg.Payload()))
consumer.Ack(msg)
atomic.AddInt32(&count, 1)

}

if err := consumer.Unsubscribe(); err != nil {
log.Fatal(err)
}
}

```

Go demo result
```shell
go build test_consumer.go
./test_consumer

...

2021/03/15 16:48:45 consume 1250000 msg
2021/03/15 16:49:45 consume 646270 msg
2021/03/15 16:50:45 consume 497119 msg
2021/03/15 16:51:45 consume 422100 msg
2021/03/15 16:52:45 consume 372000 msg
2021/03/15 16:53:45 consume 338500 msg
2021/03/15 16:54:45 consume 312500 msg
2021/03/15 16:55:45 consume 288000 msg
2021/03/15 16:56:45 consume 273500 msg
2021/03/15 16:57:45 consume 258000 msg
2021/03/15 16:58:45 consume 246000 msg
2021/03/15 16:59:45 consume 634000 msg
2021/03/15 17:00:45 consume 928500 msg
2021/03/15 17:01:45 consume 597842 msg
2021/03/15 17:02:45 consume 478458 msg
2021/03/15 17:03:45 consume 409700 msg
2021/03/15 17:04:45 consume 363500 msg
2021/03/15 17:05:45 consume 331500 msg
2021/03/15 17:06:45 consume 305300 msg
2021/03/15 17:07:45 consume 285800 msg
2021/03/15 17:08:45 consume 265900 msg
2021/03/15 17:09:45 consume 252500 msg
2021/03/15 17:10:45 consume 615500 msg
2021/03/15 17:11:45 consume 923500 msg
2021/03/15 17:12:45 consume 602100 msg
2021/03/15 17:13:45 consume 479400 msg
2021/03/15 17:14:45 consume 414920 msg
2021/03/15 17:15:45 consume 366576 msg
2021/03/15 17:16:45 consume 327500 msg
2021/03/15 17:17:45 consume 304426 msg
2021/03/15 17:18:45 consume 285570 msg
2021/03/15 17:19:45 consume 272000 msg
2021/03/15 17:20:45 consume 251900 msg
2021/03/15 17:21:45 consume 244100 msg

...
```

---

Python demo
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
import pulsar
import threading

PULSAR_URL = "my url"
TOPIC = "my_topic"
RECEIVE_MSG_TIMEOUT_MS = 3000
CONSUMER_TYPE = pulsar.ConsumerType.Shared

consume_cnt = 0

def consume():
client = pulsar.Client(PULSAR_URL)
consumer = client.subscribe(TOPIC,
subscription_name="test_consume_py" + str(time.time()),
consumer_type=pulsar.ConsumerType.Shared,
initial_position=pulsar.InitialPosition.Earliest)
global consume_cnt

while True:
try:
msg = consumer.receive(RECEIVE_MSG_TIMEOUT_MS)
consumer.acknowledge(msg)
consume_cnt += 1
except KeyboardInterrupt:
exit(-1)
# never close

class myThread(threading.Thread):
def __init__(self,):
threading.Thread.__init__(self)

def run(self):
show_num()

def show_num():
# count consume msg num every 1 min
while 1:
global consume_cnt
time.sleep(60)
print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+ " consume " + str(consume_cnt) + " msg.")
consume_cnt = 0

if __name__ == '__main__':
thread1 = myThread()
thread1.start()
consume()
thread1.join()
```

Python demo result
```shell
...
2021-03-15 17:15:22 consume 3719645 msg.
2021-03-15 17:16:22 consume 3731980 msg.
2021-03-15 17:17:22 consume 3762029 msg.
2021-03-15 17:18:22 consume 3799080 msg.
2021-03-15 17:19:22 consume 3990855 msg.
2021-03-15 17:20:22 consume 3813902 msg.
2021-03-15 17:21:22 consume 3718900 msg.
2021-03-15 17:22:22 consume 3646500 msg.
```

#### System configuration
OS: Centos
go version: 1.15
**pulsar go client version**: v0.4.0