日志服务(原SLS)新功能发布(3)--多实例协同消费库(loghub client library)

使用场景

loghub client library是对LogHub消费者提供的高级模式,解决多个消费者同时消费logstore时自动分配shard问题。
例如在stormspark streaming场景中多个消费者情况下,自动处理shard的负载均衡,消费者failover等逻辑。用户只需专注在自己业务逻辑上,而无需关心shard分配、CheckPoint、Failover等事宜。

举一个例子而言,用户需要通过storm进行流计算,启动了A、B、C 3个消费实例。在有10个shard情况下,系统会自动为A、B、C分配3、3、4个Shard进行消费。

  • 当消费实例A宕机情况下,系统会把A未消费的3个Shard中数据自动均衡B、C上,当A恢复后,会重新均衡。
  • 当添加实例D、E情况下,系统会自动进行均衡,每个实例消费2个Shard。
  • 当Shard有Merge/Split等情况下,会根据最新的Shard信息,重新均衡。
  • 当read only状态的shard消费完之后,剩余的shard会重新做负载均衡。

以上整个过程不会产生数据丢失、以及重复,用户只需在代码中做三件事情:

  1. 创建Consumer Group。
  2. 将实例名注册为Instance,并连接到Consumer Group中。
  3. 写处理日志的代码。

我们强烈建议使用loghub client library进行数据消费,这样您只需要关心怎么处理数据,而不需要关注复杂的负载均衡、消费断点保存、按序消费、消费异常处理等问题

术语简介

loghub client library中主要有4个概念,分别是consumer group、consumer、heartbeat和checkpoint,它们之间的关系如下:

日志服务(原SLS)新功能发布(3)--多实例协同消费库(loghub client library)

  • consumer group

    是logstore的子资源,拥有相同consumer group 名字的消费者共同消费同一个logstore的所有数据,这些消费者之间不会重复消费数据,一个logstore下面可以最多创建5个consumer group,不可以重名,同一个logstore下面的consumer group之间消费数据不会互相影响。consumer group有两个很重要的属性:
{
    "order":boolean,
    "timeout": integer
}

order属性表示是否按照写入时间顺序消费key相同的数据,timeout表示consumer group中消费者的超时时间,单位是秒,当一个消费者汇报心跳的时间间隔超过timeout,会被认为已经超时,服务端认为这个consumer此时已经下线了。

  • consumer

    消费者,每个consumer上会被分配若干个shard,consumer的职责就是要消费这些shard上的数据,同一个consumer group中的consumer必须不重名。
    
  • heartbeat

    消费者心跳,consumer需要定期向服务端汇报一个心跳包,用于表明自己还处于存活状态。
  • checkpoint

    消费者定期将分配给自己的shard消费到的位置保存到服务端,这样当这个shard被分配给其它消费者时,从服务端可以获取shard的消费断点,接着从断点继续消费数据。
    

实现原理

通信协议

定义如下消费者和服务端的通信接口:

/*
    创建consumr group,inOrder表示是否希望key相同的数据能够按照写入的时间顺序被消费,timeoutInSec表示consumer的心跳超时时间,超过timeoutInSec没有汇报心跳的consumer会被认为已经下线了,建议取值20s左右。
*/
boolean CreateConsumerGroup(
            String project, 
            String logstore, 
            String consumerGroupName, 
            boolean inOrder,
            int timeoutInSec);
/*
    删除consumer group
*/
boolean DeleteConsumerGroup(
            String project, 
            String logStore, 
            String consumerGroup);
/*
    列出logstore下所有的consumer group,包括每个consumer group的order、timeout属性
*/
List<ConsumerGroup> ListConsumerGroup(
            String project, 
            String logStore);
/*
    更新consumer group的属性,consumer group的名字不可以更新,如果将inOrder由true更新为false,那么当前所有未开始消费的shard将会被分配至各个consumer,如果inOrder由false更新为true,对于当前正在消费的shard不会生效,也就是说即使shard之间有父子关系,由于它们都已经被消费了,所以顺序对他们而言没有意义,但是inOrder属性更新之后,shard分裂和合并产生的新的shard将会被顺序消费。
*/
boolean UpdateConsumerGroup(
            String project,
            String logStore,
            String consumerGroup,
            boolean inOrder,
            int timeoutInSec
            );
/*
    更新shard消费到的位置,只有当参数中的shard当前由consumer持有的情况下才能更新成功
*/
boolean UpdateCheckPoint(
            String project, 
            String logStore, 
            String consumerGroup, 
            String consumer,
            int shard,
            String checkpoint);
/*
    更新shard消费到的位置,无论如何都可以更新成功
*/
boolean UpdateCheckPoint(
            String project, 
            String logStore, 
            String consumerGroup, 
            int shard,
            String checkpoint);
/*
    获取consumer group中shard的消费断点
*/
String GetCheckPoint(
            String project,
            String logStore,
            String consumerGroup,
            int shard);
/*
    获取consumer group中所有shard的消费断点,返回结果是一个断点的List,具体请参考sdk
*/
List<ShardCheckPoint> GetCheckPoint(
            String project,
            String logStore,
            String consumerGroup);
/*
    将当前consumer持有的shard汇报给服务端,服务端返回确认包,里面包含若干shard,假设持有的shard集合为A,返回的确认shard集合为B,A和B的差集(A-B)表示consumer应该放弃消费的shard,consumer应该尽快将(A-B)中的shard的消费断点保存到服务端,并放弃消费(A-B)中的shard,(B-A)表示consumer可以增持的shard,consumer从服务端获取(B-A)中shard的checkpoint。理想状态下下一次心跳汇报给服务端的shard集合就是B,但是如果consumer不想(没来得及)放弃A-B中的某些shard,需要将这些shard一并汇报给服务端。Heartbeat除了上面有消费负载均衡的功能以外,还用于告知服务端自己处于存活状态,不要将自己从consumer group中删除。一旦heartbeat超时,服务端就会将consumer从consumer group中删除,并将分配给其的shard重新分配给别的consumer。
*/
List<Integer> HeartBeat(
            String project,
            String logStore,
            String consumerGroup,
            String consumer,
            ArrayList<Integer> shards
            )

有限状态自动机

服务端对consumer group中每个shard都会维护一个有限状态自动机,共有五种状态,分别是already_alloc、not_alloc、wait、transfer、over,每种状态的含义如下:

already_alloc:该shard已经被某个consumer持有并消费。

not_alloc:该shard可以消费,但是尚未被任何consumer持有。

wait:该shard当前不可以消费,需要等待其祖先shard消费完。

transfer:将该shard转交给另一个消费者消费的过渡状态。只有当当前消费者在Heartbeat中放弃消费该shard,才能将该shard转交出去,因此需要这个过渡状态。

over:该shard的数据已经消费完。

wait状态需要重点说明下,假设某个时刻数据仓库所有shard的关系如下图:

日志服务(原SLS)新功能发布(3)--多实例协同消费库(loghub client library)

初始时有3个shard 0、1、2,每个shard下面的区间表示关联的hash值的集合,这里为了简单用数值型表示hash key,此时hash值为7的数据会被写入shard 0,随后0 split成3和4,shard 0变成read_only状态,shard 3和4变成read_write状态,这个时候hash值是7的数据不会再被写入shard 0,而是写入shard 4,再接着shard 4和5 merge成6之后,hash值为7的数据只会写入shard 6。

如果要顺序消费hash值为7的数据,必须保证在shard 0被消费完之前shard 4不应该被任何消费者消费,同理,shard 6不应该在shard 4中数据消费完之前被消费,我们把shard 0、4认为是shard 6的祖先,shard 6称为shard 0和4的后代,某个shard可以被消费当且仅当其祖先shard中的数据被消费完。基于这个原因,引入wait状态表示该shard当前不可以被消费。

状态转移图如下:

日志服务(原SLS)新功能发布(3)--多实例协同消费库(loghub client library)

图中每个状态用首字母缩写表示,每条边对应含义如下:

1 表示shard的起始状态只能是wait或者not_alloc。

2 表示该shard的祖先shard的数据已经被消费完,可以开始消费当前shard的数据了。

3 表示该shard被分配给了某一个消费者。

4 表示消费该shard的consumer心跳超时了,回收其持有的shard。

5 表示持有该shard的消费者持有的shard总数太多,不满足任意两个消费者持有shard数量之差的绝对值小于等于1,所以将该shard转移给别的需要的消费者消费,这时会将等待消费的消费者(next consumer)和这个shard关联起来。

6 表示收到持有该shard的consumer放弃消费的Heartbeat,并将该shard转移给关联的next consumer持有。发生该转移还有一种可能是该shard的next consumer超时,继续由持有该shard的消费者持有该shard。

7 表示当强制更新over状态的shard的消费断点到某个非数据结束位置时,该shard恢复可消费状态。执行这种更新操作要特别当心,因为该shard的后代shard可能已经被消费了,很可能导致数据无法按照hash key的顺序消费。

8 表示read_only状态的shard数据被消费完了。

9 表示持有该shard的消费者Heartbeat超时,回收该shard到not_alloc状态。

这里要注意以下几点:

  • shard 处于transfer状态时,服务端收到持有该shard的消费者的Heartbeat时,返回的确认shard集合中不会包含该shard,确认shard集合中只会包含持有者是该consumer并且shard状态是already_alloc的shard。
  • 消费者调用UpdateCheckpoint更新消费断点,如果是read_only状态的shard要检查该checkpoint是否是shard的结尾,如果是就需要将shard状态转移成over。
  • 5 标识的转移是保证任意两个消费者持有shard数量之差的绝对值小于等于1的基础,当发现有consumer持有的shard数量不满足该条件时,从shard持有数量多的消费者那里剥夺一些shard,分配给持有数量少的consumer,这个过程称为消费负载均衡。这些将要易手的shard需要设置成transfer状态,以等待持有者heartbeat中确认放弃,这主要是为了让持有者收到放弃消息时将消费断点保存到服务端,从而易手之后,新的consumer可以从服务端获取该shard的消费断点。
  • 新的shard加入时,只能由转移1进入,其状态为wait当该shard有祖先shard,并且其祖先没有消费完,否则状态为not_alloc。
  • 消费的负载均衡只会考虑not_alloc、already_alloc、transfer状态的shard,wait和over状态的shard由于不满足消费条件,所以不会被分配给任何consumer。

如何使用loghub client library

  • 实现loghub client library中的两个接口类:

    • ILogHubProcessor // 每个shard对应一个实例,每个实例只消费特定shard的数据。
    • ILogHubProcessorFactory // 负责生产实现ILogHubProcessor接口实例。
  • 填写参数配置。
  • 启动一个或多个client worker实例。

使用sample

main函数

    public static void main(String args[]) 
    {
        LogHubConfig config = new LogHubConfig(...);
                
        ClientWorker worker = new ClientWorker(new SampleLogHubProcessorFactory(), config);
        
        Thread thread = new Thread(worker);
        //thread运行之后,client worker会自动运行,ClientWorker扩展了Runnable接口。
        thread.start();
                Thread.sleep(60 * 60 * 1000);
        //调用worker的shutdown函数,退出消费实例,关联的线程也会自动停止。
        worker.shutdown();
        //ClientWorker运行过程中会生成多个异步的Task,shutdown之后最好等待还在执行的Task安全退出,建议30s。
        Thread.sleep(30 * 1000);
    }

ILogHubProcessor、ILogHubProcessorFactory 实现sample

  • 各个shard对应的消费实例类,实际开发过程中用户主要需要关注数据消费逻辑,同一个ClientWorker实例是串行消费数据的,只会产生一个ILogHubProcessor实例,ClientWorker退出的时候会调用ILogHubProcessor的shutdown函数。
public class SampleLogHubProcessor implements ILogHubProcessor 
{
    private int mShardId;
    // 记录上次持久化check point的时间
    private long mLastCheckTime = 0; 
    
    public void initialize(int shardId) 
    {
        mShardId = shardId;
    }

    // 消费数据的主逻辑
    public String process(List<LogGroupData> logGroups,
            ILogHubCheckPointTracker checkPointTracker) 
    {
        for (LogGroupData group : logGroups) 
        {
            List<LogItem> items = group.GetAllLogs();
            for (LogItem item : items) 
            {
                // 打印loggroup中的数据
                System.out.println("shard_id:" + mShardId + " " + item.ToJsonString());
            }
        }
        long curTime = System.currentTimeMillis();
        // 每隔60秒,写一次check point到服务端,如果60秒内,worker crash,
        // 新启动的worker会从上一个checkpoint其消费数据,有可能有重复数据
        if (curTime - mLastCheckTime >  60 * 1000) 
        {
            try 
            {
                checkPointTracker.saveCheckPoint(true);
            } 
            catch (LogHubCheckPointException e) 
            {
                e.printStackTrace();
            }
            mLastCheckTime = curTime;
        } 
        else 
        {
            try 
            {
                checkPointTracker.saveCheckPoint(false);
            } 
            catch (LogHubCheckPointException e) 
            {
                e.printStackTrace();
            }
        }
        // 返回空表示正常处理数据, 如果需要回滚到上个check point的点进行重试的话,可以return checkPointTracker.getCheckpoint()
        return null;  
    }
    // 当worker退出的时候,会调用该函数,用户可以在此处做些清理工作。
    public void shutdown(ILogHubCheckPointTracker checkPointTracker) 
    {
        //将消费断点保存到服务端。
        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
    }
}
  • 生成 ILogHubProcessor的工厂类 :
public class SampleLogHubProcessorFactory implements ILogHubProcessorFactory 
{
    public ILogHubProcessor generatorProcessor()
    {   
        // 生成一个消费实例
        return new SampleLogHubProcessor();
    }
}

配置说明:

public class LogHubConfig 
{
    //worker默认的拉取数据的时间间隔
    public static final long DEFAULT_DATA_FETCH_INTERVAL_MS = 200;
    //consumer group的名字
    private String mConsumerGroupName;
    //consumer的名字,必须确保同一个consumer group下面的各个consumer不重名
    private String mWorkerInstanceName;
    //loghub数据接口地址
    private String mLogHubEndPoint;
    //项目名称
    private String mProject;
    //日志库名称
    private String mLogStore;
    //云账号的access key id
    private String mAccessId;
    //云账号的access key
    private String mAccessKey;
    //用于指出在服务端没有记录shard的checkpoint的情况下应该从什么位置消费shard,取值可以是[BEGIN_CURSOR, END_CURSOR, SPECIAL_TIMER_CURSOR]中的一个
    private LogHubCursorPosition mCursorPosition;
    //当mCursorPosition取值为SPECIAL_TIMER_CURSOR时,指定消费时间,单位是秒。
    private int  mLoghubCursorStartTime = 0;
    // 轮询获取loghub数据的时间间隔,间隔越小,抓取越快,单位是毫秒,默认是DEFAULT_DATA_FETCH_INTERVAL_MS,建议时间间隔200ms以上。
    private long mDataFetchIntervalMillis;
    // worker向服务端汇报心跳的时间间隔,单位是毫秒,建议取值10000ms。
    private long mHeartBeatIntervalMillis;
    //是否按序消费
    private boolean mConsumeInOrder; 
}

Maven配置

<dependency>
  <groupId>com.aliyun.openservices</groupId>
  <artifactId>loghub-client-lib</artifactId>
  <version>0.6.4</version>
</dependency>

Storm, Spark Streaming Library

常见问题&注意事项

  • LogHubConfig 中 consumerGroupName表一个消费组,consumerGroupName相同的consumer分摊消费logstore中的shard,同一个consumerGroupName中的consumer,通过workerInstance name进行区分。
假设logstore中有shard 0 ~ shard 3 这4个shard。
有3个worker,其consumerGroupName和workerinstance name分别是 : 
<consumer_group_name_1 , worker_A>,
<consumer_group_name_1 , worker_B>,
<consumer_group_name_2 , worker_C>
则,这些worker和shard的分配关系是:
<consumer_group_name_1 , worker_A>: shard_0, shard_1
<consumer_group_name_1 , worker_B>: shard_2, shard_3
<consumer_group_name_2 , worker_C>: shard_0, shard_1, shard_2, shard_3  # group name不同的worker相互不影响
  • 确保实现的ILogHubProcessor process()接口每次都能顺利执行,并退出,这点很重要。
  • ILogHubCheckPointTracker的saveCheckPoint()接口,无论传递的参数是true,还是false,都表示当前处理的数据已经完成,参数为true,则立刻持久化至服务端,false则每隔60秒同步一次到服务端。
  • 如果LogHubConfig中配置的是子用户的accessKeyId、accessKey,需要在RAM中进行以下授权,详细内容请参考API文档

    |Action|Resource|
log:GetCursorOrData acs:log:{regionName}:{projectOwnerAliUid}:project/{projectName}/logstore/{logstoreName}
log:CreateConsumerGroup acs:log:{regionName}:{projectOwnerAliUid}:project/{projectName}/logstore/{logstoreName}/consumergroup/*
log:ListConsumerGroup acs:log:{regionName}:{projectOwnerAliUid}:project/{projectName}/logstore/{logstoreName}/consumergroup/*
log:ConsumerGroupUpdateCheckPoint acs:log:{regionName}:{projectOwnerAliUid}:project/{projectName}/logstore/{logstoreName}/consumergroup/{consumerGroupName}
log:ConsumerGroupHeartBeat acs:log:{regionName}:{projectOwnerAliUid}:project/{projectName}/logstore/{logstoreName}/consumergroup/{consumerGroupName}
log:GetConsumerGroupCheckPoint acs:log:{regionName}:{projectOwnerAliUid}:project/{projectName}/logstore/{logstoreName}/consumergroup/{consumerGroupName}
上一篇:LINUX下Git连接到Github 并 实现简单仓库例子


下一篇:[IOS]陀螺仪和加速仪编程