前段時間在 pulsar-client-go 社區里看到這么一個 issue：

圖片

import "github.com/apache/pulsar-client-go/pulsar"

client, err := pulsar.NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
if err != nil {
    log.Fatal(err)
}
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:             "persistent://public/default/mq-topic-1",
    SubscriptionName:  "sub-1",
    Type:              pulsar.Shared,
    ReceiverQueueSize: 0,
})
if err != nil {
    log.Fatal(err)
}


// 小于等于 0 時會設置為 1000
const (  
    defaultReceiverQueueSize = 1000  
)
if options.ReceiverQueueSize <= 0 {  
    options.ReceiverQueueSize = defaultReceiverQueueSize  
}

他發現手動將 pulsar-client-go 客戶端的 ReceiverQueueSize 設置為 0 的時候，客戶端在初始化時會再將其調整為 1000.

if options.ReceiverQueueSize < 0 {  
    options.ReceiverQueueSize = defaultReceiverQueueSize  
}

而如果手動將源碼修改為可以設置為 0 時，卻不能正常消費，消費者會一直處于 waiting 狀態，獲取不到任何數據。

經過我的排查發現是 Pulsar 的  Go  客戶端缺少了一個 ZeroQueueConsumerImpl的實現類，這個類主要用于可以精細控制消費邏輯。

If you'd like to have tight control over message dispatching across consumers, set the consumers' receiver queue size very low (potentially even to 0 if necessary). Each consumer has a receiver queue that determines how many messages the consumer attempts to fetch at a time. For example, a receiver queue of 1000 (the default) means that the consumer attempts to process 1000 messages from the topic's backlog upon connection. Setting the receiver queue to 0 essentially means ensuring that each consumer is only doing one thing at a time.

https://pulsar.apache.org/docs/next/cookbooks-message-queue/#client-configuration-changes

正如官方文檔里提到的那樣，可以將 ReceiverQueueSize 設置為 0；這樣消費者就可以一條條的消費數據，而不會將消息堆積在客戶端隊列里。

客戶端消費邏輯

借此機會需要再回顧下 pulsar 客戶端的消費邏輯，這樣才能理解 ReceiverQueueSize 的作用以及如何在 pulsar-client-go 如何實現這個 ZeroQueueConsumerImpl。

Pulsar 客戶端的消費模式是基于推拉結合的：

圖片

如這張圖所描述的流程，消費者在啟動的時候會主動向服務端發送一個 Flow 的命令，告訴服務端需要下發多少條消息給客戶端。

同時會使用剛才的那個 ReceiverQueueSize參數作為內部隊列的大小，將客戶端下發的消息存儲在內部隊列里。

然后在調用 receive 函數的時候會直接從這個隊列里獲取數據。

圖片

圖片

每次消費成功后都會將內部的一個 AvailablePermit+1，直到大于 MaxReceiveQueueSize / 2 就會再次向 broker 發送 flow 命令，告訴 broker 再次下發消息。

所以這里有一個很關鍵的事件：就是向 broker 發送 flow 命令，這樣才會有新的消息下發給客戶端。

之前經常都會有研發同學讓我排查無法消費的問題，最終定位到的原因幾乎都是消費緩慢，導致這里的 AvailablePermit 沒有增長，從而也就不會觸發 broker 給客戶端推送新的消息。

看到的現象就是消費非常緩慢。

ZeroQueueConsumerImpl 原理

下面來看看 ZeroQueueConsumerImpl 是如何實現隊列大小為 0 依然是可以消費的。

圖片

在構建 consumer 的時候，就會根據隊列大小從而來創建普通消費者還是 ZeroQueueConsumerImpl 消費者。

@Override  
protected CompletableFuture<Message<T>> internalReceiveAsync() {  
    CompletableFuture<Message<T>> future = super.internalReceiveAsync();  
    if (!future.isDone()) {  
        // We expect the message to be not in the queue yet  
        increaseAvailablePermits(cnx());  
    }  
    return future;  
}

這是 ZeroQueueConsumerImpl 重寫的一個消費函數，其中關鍵的就是 increaseAvailablePermits(cnx());.

void increaseAvailablePermits(ClientCnx currentCnx) {
        increaseAvailablePermits(currentCnx, 1);
    }

    protected void increaseAvailablePermits(ClientCnx currentCnx, int delta) {
        int available = AVAILABLE_PERMITS_UPDATER.addAndGet(this, delta);
        while (available >= getCurrentReceiverQueueSize() / 2 && !paused) {
            if (AVAILABLE_PERMITS_UPDATER.compareAndSet(this, available, 0)) {
                sendFlowPermitsToBroker(currentCnx, available);
                break;
            } else {
                available = AVAILABLE_PERMITS_UPDATER.get(this);
            }
        }
    }

從源碼里可以得知這里的邏輯就是將 AvailablePermit 自增，達到閾值后請求 broker 下發消息。

因為在 ZeroQueueConsumerImpl 中隊列大小為 0，所以 available >= getCurrentReceiverQueueSize() / 2永遠都會為 true。

也就是說每消費一條消息都會請求 broker 讓它再下發一條消息，這樣就達到了每一條消息都精確控制的效果。

pulsar-client-go 中的實現

為了在 pulsar-client-go 實現這個需求，我提交了一個 PR 來解決這個問題。

其實從上面的分析已經得知為啥手動將 ReceiverQueueSize 設置為 0 無法消費消息了。

根本原因還是在初始化的時候優于隊列為 0，導致不會給 broker 發送 flow 命令，這樣就不會有消息推送到客戶端，也就無法消費到數據了。

所以我們依然得參考 Java 的 ZeroQueueConsumerImpl 在每次消費的時候都手動增加  availablePermits。

為此我也新增了一個消費者 zeroQueueConsumer。

// EnableZeroQueueConsumer, if enabled, the ReceiverQueueSize will be 0.  
// Notice: only non-partitioned topic is supported.  
// Default is false.  
EnableZeroQueueConsumer bool

consumer, err := client.Subscribe(ConsumerOptions{  
    Topic:                   topicName,  
    SubscriptionName:        "sub-1",  
    Type:                    Shared,  
    NackRedeliveryDelay:     1 * time.Second,  
    EnableZeroQueueConsumer: true,  
})

if options.EnableZeroQueueConsumer {  
    options.ReceiverQueueSize = 0  
}

在創建消費者的時候需要指定是否開啟 ZeroQueueConsumer，當開啟后會手動將 ReceiverQueueSize 設置為 0.

// 可以設置默認值。
private int receiverQueueSize = 1000;

在 Go 中無法像 Java 那樣在結構體初始化化的時候就指定默認值，再加上 Go 的 int 類型具備零值（也就是0），所以無法區分出 ReceiverQueueSize=0 是用戶主動設置的，還是沒有傳入這個參數使用的零值。

所以才需要新增一個參數來手動區分是否使用 ZeroQueueConsumer。

圖片

之后在創建 consumer 的時候進行判斷，只有使用的是單分區的 topic 并且開啟了 EnableZeroQueueConsumer 才能創建  zeroQueueConsumer。

圖片

使用 PARTITIONED_METADATA 命令可以讓 broker 返回分區數量。

func (z *zeroQueueConsumer) Receive(ctx context.Context) (Message, error) {
 if state := z.pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
  z.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
  return nil, errors.New("consumer state is closed")
 }
 z.Lock()
 defer z.Unlock()
 z.pc.availablePermits.inc()
 for {
  select {
  case <-z.closeCh:
   return nil, newError(ConsumerClosed, "consumer closed")
  case cm, ok := <-z.messageCh:
   if !ok {
    return nil, newError(ConsumerClosed, "consumer closed")
   }
   return cm.Message, nil
  case <-ctx.Done():
   return nil, ctx.Err()
  }
 }

}

其中的關鍵代碼：z.pc.availablePermits.inc()

消費時的邏輯其實和 Java 的 ZeroQueueConsumerImpl 邏輯保持了一致，也是每消費一條數據之前就增加一次 availablePermits。

pulsar-client-go 的運行原理與 Java 客戶端的類似，也是將消息存放在了一個內部隊列里，所以每次消費消息只需要從這個隊列 messageCh 里獲取即可。

值得注意的是， pulsar-client-go 版本的 zeroQueueConsumer 就不支持直接讀取內部的隊列了。

func (z *zeroQueueConsumer) Chan() <-chan ConsumerMessage {  
    panic("zeroQueueConsumer cannot support Chan method")  
}

會直接 panic，因為直接消費 channel 在客戶端層面就沒法幫用戶主動發送 flow 命令了，所以這個功能就只能屏蔽掉了，只可以主動的 receive 消息。

圖片

許久之前我也畫過一個關于 pulsar client 的消費流程圖，后續考慮會再寫一篇關于 pulsar client 的原理分析文章。

參考鏈接：

• https://github.com/apache/pulsar-client-go/issues/1223
• https://cloud.tencent.com/developer/article/2307608
• https://pulsar.apache.org/docs/next/cookbooks-message-queue/#client-configuration-changes
• https://github.com/apache/pulsar-client-go/pull/1225