91      阿裏雲  技術社區[雲棲]

LogHub-協同消費組__Getting-Started_日誌服務-阿裏雲

協同消費組（ConsumerGroup）是實時消費數據高級模式，能夠提供多個消費實例對日誌庫消費自動負載均衡。Spark Streaming、Storm都以ConsumerGroup作為基礎模式。

通過控製查看消費進度

1. 選擇project進入協同消費功能頁麵，選擇日誌庫（LogStore）後即可查看目前是否啟用協同消費功能
2. 選擇指定ConsumerGroup之後，點擊“消費狀態”鏈接，即可查看當前每個Shard消費數據的進度

如上圖所示，頁麵上展示該日誌庫包含5個Shard，對應5個消費者，其中每個消費者最近消費的數據時間如第二列顯示。通過消費數據時間可以判斷出目前數據處理是否能滿足數據產生速度，如果已經嚴重落後於當前時間（即數據消費速率小於數據產生速率），可以考慮增加消費者數目。

通過API/SDK查看消費進度

以Java SDK作為例子，演示如何通過API獲得消費狀態：

package test;
import java.util.ArrayList;
import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.Consts.CursorMode;
import com.aliyun.openservices.log.common.ConsumerGroup;
import com.aliyun.openservices.log.common.ConsumerGroupShardCheckPoint;
import com.aliyun.openservices.log.exception.LogException;
public class ConsumerGroupTest {
    static String endpoint = "";
    static String project = "";
    static String logstore = "";
    static String accesskeyId = "";
    static String accesskey = "";
    public static void main(String[] args) throws LogException {
        Client client = new Client(endpoint, accesskeyId, accesskey);
        //獲取這個logstore下的所有consumer group，可能不存在，此時consumerGroups的長度是0
        ArrayList<ConsumerGroup>  consumerGroups;
        try{
            consumerGroups = client.ListConsumerGroup(project, logstore).GetConsumerGroups();
        }
        catch(LogException e){
            if(e.GetErrorCode() == "LogStoreNotExist")
                System.out.println("this logstore does not have any consumer group");
            else{
                //internal server error branch
            }
            return;
        }
        for(ConsumerGroup c: consumerGroups){
            //打印consumer group的屬性，包括名稱、心跳超時時間、是否按序消費
            System.out.println("名稱: " + c.getConsumerGroupName());
            System.out.println("心跳超時時間: " + c.getTimeout());
            System.out.println("按序消費: " + c.isInOrder());
            for(ConsumerGroupShardCheckPoint cp: client.GetCheckPoint(project, logstore, c.getConsumerGroupName()).GetCheckPoints()){
                System.out.println("shard: " + cp.getShard());
                //請格式化下，這個時間返回精確到毫秒的時間，長整型
                System.out.println("最後一次消費數據的時間: " + cp.getUpdateTime());
                System.out.println("消費者名稱: " + cp.getConsumer());
                String consumerPrg = "";
                if(cp.getCheckPoint().isEmpty())
                    consumerPrg = "尚未開始消費";
                else{
                    //unix時間戳，單位是秒，輸出的時候請注意格式化
                    try{
                        int prg = client.GetPrevCursorTime(project, logstore, cp.getShard(), cp.getCheckPoint()).GetCursorTime();
                        consumerPrg = "" + prg;
                    }
                    catch(LogException e){
                        if(e.GetErrorCode() == "InvalidCursor")
                            consumerPrg = "非法，前一次消費時刻已經超出了logstore中數據的生命周期";
                        else{
                            //internal server error
                            throw e;
                        }
                    }
                }
                System.out.println("消費進度: " + consumerPrg);
                String endCursor = client.GetCursor(project, logstore, cp.getShard(), CursorMode.END).GetCursor();
                int endPrg = 0;
                try{
                    endPrg = client.GetPrevCursorTime(project, logstore, cp.getShard(), endCursor).GetCursorTime();
                }
                catch(LogException e){
                    //do nothing
                }
                //unix時間戳，單位是秒，輸出的時候請注意格式化
                System.out.println("最後一條數據到達時刻: " + endPrg);
            }
        }
    }
}