Simple Log Service:コンシューマーグループ - 使用法

• コンシューマーグループ

  コンシューマーグループは複数のコンシューマーで構成されます。コンシューマーグループ内のコンシューマーは、同じ Logstore のデータを読み込みます。コンシューマーごとに別々のデータが読み込まれます。

• コンシューマー

  コンシューマーは、データを読み込む、コンシューマーグループを構成する単位です。コンシューマー名は、コンシューマーグループ内で一意である必要があります。

• 各シャードに割り当てることのできるコンシューマーは 1 つのみ
• 各コンシューマーには、複数のシャードを紐づけることができる

public class Main {
  // Log Service のドメイン名を指定
  private static String sEndpoint = "cn-hangzhou.log.aliyuncs.com";
  // Log Service のプロジェクト名を指定
  private static String sProject = "ali-cn-hangzhou-sls-admin";
  // Log Service の Logstore 名を指定
  private static String sLogstore = "sls_operation_log";
  // コンシューマーグループ名を指定
  private static String sConsumerGroup = "consumerGroupX";
  // データ読み込みの AccessKey を指定
  private static String sAccessKeyId = "";
  private static String sAccessKey = "";
  public static void main(String []args) throws LogHubClientWorkerException, InterruptedException
  {
       // ２ つ目のパラメータはコンシューマー名です。コンシューマーグループ内のコンシューマー名はコンシューマーグループ内で一意である必要があります。ただし、コンシューマーグループ名は重複しても構いません。各コンシューマーは複数のマシンの複数の処理を負荷分散方式で読み込み開始します。 コンシューマー名はマシンの IP アドレス順に並べられます。9 番目には、maxFetchLogGroupSize パラメータを指定します (maxFetchLogGroupSize は、Log Service が 1 回に取得する Logstore の数ですが、初期値をそのままご使用ください。変更する場合には、1 以上 1000 以下にします)。
      LogHubConfig config = new LogHubConfig(sConsumerGroup, "consumer_1", sEndpoint, sProject, sLogstore, sAccessKeyId, sAccessKey, LogHubConfig.ConsumePosition.BEGIN_CURSOR);
      ClientWorker worker = new ClientWorker(new SampleLogHubProcessorFactory(), config);
        Thread thread = new Thread(worker);
        //スレッドの実行が開始し、実行可能な API を extend すると ClientWorker が自動的に起動されます。
       thread.start();
       Thread.sleep(60 * 60 * 1000);
       //worker の Shutdown 関数を呼び出して読み込みインスタンスを 終了します。紐づいているスレッドは自動停止します。
       worker.shutdown();
       //ClientWorker 実行中に複数の非同期タスクが生成されます。シャットダウンしたら、実行タスクが終了するまで 30 秒待機することを推奨します。
       Thread.sleep(30 * 1000);
  }
}

public class SampleLogHubProcessor implements ILogHubProcessor 
{
  private int mShardId;
  // 最後に残ったチェックポイント時間を記録します。
  private long mLastCheckTime = 0; 
  public void initialize(int shardId) 
  {
      mShardId = shardId;
  }
  // データ読み込みの主要部。例外はすべてキャッチされますが、キャッチされた例外は返されません。
  public String process(List<LogGroupData> logGroups,
          ILogHubCheckPointTracker checkPointTracker) 
  {
      // Write checkpoint to30 秒ごとに Log Service にチェックポイントを書き込みます。30 秒以内に worker がクラッシュすると、新たにクラッシュすると、新たに開始した worker が最終チェックポイントよりデータの読み込みを開始します。わずかに重複データが含まれる場合もあります。
      for(LogGroupData logGroup: logGroups){
          FastLogGroup flg = logGroup.GetFastLogGroup();
          System.out.println(String.format("\tcategory\t:\t%s\n\tsource\t:\t%s\n\ttopic\t:\t%s\n\tmachineUUID\t:\t%s",
                  flg.getCategory(), flg.getSource(), flg.getTopic(), flg.getMachineUUID()));
          System.out.println("Tags");
          for (int tagIdx = 0; tagIdx < flg.getLogTagsCount(); ++tagIdx) {
              FastLogTag logtag = flg.getLogTags(tagIdx);
              System.out.println(String.format("\t%s\t:\t%s", logtag.getKey(), logtag.getValue()));
          }
          for (int lIdx = 0; lIdx < flg.getLogsCount(); ++lIdx) {
              FastLog log = flg.getLogs(lIdx);
              System.out.println("--------\nLog: " + lIdx + ", time: " + log.getTime() + ", GetContentCount: " + log.getContentsCount());
              for (int cIdx = 0; cIdx < log.getContentsCount(); ++cIdx) {
                  FastLogContent content = log.getContents(cIdx);
                  System.out.println(content.getKey() + "\t:\t" + content.getValue());
              }
          }
      }
      long curTime = System.currentTimeMillis();
      // 30 秒ごとに Log Service にチェックポイントを書き込みます。30 秒以内に worker がクラッシュする場合、
      // 新たに起動した worker が最終チェックポイントから、データ読み込みを開始します。わずかに重複データが含まれる可能性があります。
      if (curTime - mLastCheckTime > 30 * 1000) 
      {
          try 
          {
              //パラメータが true の場合、Log Service のチェックポイントが更新されます。パラメータが false の場合、チェックポイントはローカルマシンにキャッシュされ、Log Service にデフォルトで  60 秒ごとに Log Service のチェックポイントが更新されます。
              checkPointTracker.saveCheckPoint(true);
          } 
          catch (LogHubCheckPointException e) 
          {
              e.printStackTrace();
          }
          mLastCheckTime = curTime;
      } 
      return null;  
  }
  // 終了する際、worker は本関数を呼び出します。ここできれいにします (cleanup)。
  public void shutdown(ILogHubCheckPointTracker checkPointTracker) 
  {
      //Log Service の読み込み中断位置の保存
      try {
          checkPointTracker.saveCheckPoint(true);
      } catch (LogHubCheckPointException e) {
          e.printStackTrace();
      }
  }
}
class SampleLogHubProcessorFactory implements ILogHubProcessorFactory 
{
  public ILogHubProcessor generatorProcessor()
  {   
      // 読み込みインスタンスの生成
      return new SampleLogHubProcessor();
  }
}

[WARN ] 2018-03-14 12:01:52,747 method:com.aliyun.openservices.loghub.client.LogHubConsumer.sampleLogError(LogHubConsumer.java:159)
com.aliyun.openservices.log.exception.LogException: Invalid loggroup count, (0,1000]

log4j.rootLogger = info,stdout
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

1. コンソールでコンシューマーグループのステータスを表示
2. CloudMonitor でコンシューマーグループの遅延を表示し、アラームを設定