a级毛片高清免费视频大全,久久人人97超碰超国产,中国熟妇牲交视频

安裝

安裝要求說明

DataHub C++ SDK目前必須使用GCC 4.9.2編譯，使用前請檢查編譯環境是否適合，確認檢查通過后再安裝使用。

SDK下載

2.13.0版本
2.14.0版本
2.15.0版本
2.23.0版本
2.25.0版本

初始化

用戶可以使用阿里云認證賬號訪問DataHub，并需要提供云賬號AccessId和AccessKey，同時需要提供訪問DataHub的服務地址。以下代碼用于使用域名列表新建DataHubClient:

     /* Configuration */
    Account account;
    account.id = "";
    account.key = "=";

    std::string projectName = "test_project";
    std::string topicName = "test_cpp";
    std::string comment = "test";

    std::string endpoint = "";
    Configuration conf(account, endpoint);

    /* Datahub Client */
    DatahubClient client(conf);

Project操作

項目（Project）是DataHub數據的基本組織單元,下面包含多個Topic。值得注意的是，DataHub的項目空間與MaxCompute的項目空間是相互獨立的。用戶在MaxCompute中創建的項目不能復用于DataHub，需要獨立創建。

創建project

void CreateProject()
{
    std::string projectName = "";
    std::string comment = "";
    try
    {
        client.CreateProject(projectName, comment);
    }
    catch(const DatahubException& e)
    {
        std::cerr << "Create project fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
}

刪除Project

void DeleteProject()
{
    std::string projectName = "";
    try
    {
        client.DeleteProject(projectName);
    }
    catch(const DatahubException& e)
    {
        std::cerr << "Delete project fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
}

更新Project

void UpdateProject()
{
  std::string projectName = "";
  std::string comment = "";
  try
  {
      client.UpdateProject(projectName,comment);
  }
  catch(const DatahubException& e)
  {
      std::cerr << "Update project fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
  }
}

獲取Project列表

void ListProject()
{
  try
  {
      const ListProjectResult& listProjectResult = client.ListProject();
      std::cout<<listProjectResult.GetProjectNames().size()<<std::endl;
  }
  catch(const DatahubException& e)
  {
      std::cerr << "List project fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
  }
}

查詢Project信息

void GetProject()
{
  std::string projectName = "";
  try
  {
     const GetProjectResult& projectResult = client.GetProject(projectName);
     std::cout<<projectResult.GetProject()<<std::endl;
     std::cout<<projectResult.GetComment()<<std::endl;
     std::cout<<projectResult.GetCreator()<<std::endl;
     std::cout<<projectResult.GetCreateTime()<<std::endl;
     std::cout<<projectResult.GetLastModifyTime()<<std::endl;    
  }
  catch(const DatahubException& e)
  {
      std::cerr << "get project fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
  }
}

Topic操作

Topic是 DataHub 訂閱和發布的最小單位，用戶可以用Topic來表示一類或者一種流數據。目前支持Tuple與Blob兩種類型：

Blob類型Topic支持寫入一塊二進制數據作為一個Record
Tuple類型的Topic支持類似于數據庫的記錄的數據，每條記錄包含多個列，需要指定Record Schema，因為網絡傳輸中，數據都是以字符串的形式發送，需要schema來轉換成對應的類型。目前支持以下幾種數據類型：

類型	含義	值域
BIGINT	8字節有符號整型	-9223372036854775807 ~ 9223372036854775807
DOUBLE	8字節雙精度浮點數	-1.0 _10^308 ~ 1.0 _10^308
BOOLEAN	布爾類型	True/False或true/false或0/1
TIMESTAMP	時間戳類型	表示到微秒的時間戳類型
STRING	字符串，只支持UTF-8編碼	單個STRING列最長允許2MB
TINYINT	單字節整型	-128 ~127
SMALLINT	雙字節整型	-32768 ~ 32767
INTEGER	4字節整型	-2147483648 ~ 2147483647
FLOAT	4字節單精度浮點數	-3.40292347_10^38 ~ 3.40292347_10^38

創建Tuple Topic

void CreateTupleTopic(){
    RecordSchema schema;
    std::string fieldName1 = "a";
    std::string fieldName2 = "b";
    std::string fieldName3 = "c";
    std::string fieldComment1 = "field1 comment";
    std::string fieldComment2 = "field2 comment";
    std::string fieldComment3 = "field3 comment";
    schema.AddField(Field(fieldName1/*Fieldname*/, BIGINT, true, fieldComment1));
    schema.AddField(Field(fieldName2/*Fieldname*/, DOUBLE, true, fieldComment2));
    schema.AddField(Field(fieldName3/*Fieldname*/, STRING, true, fieldComment3));

    /* Create Topic */
    int shardCount = 3;
    int lifeCycle = 7;
    RecordType type = TUPLE;
    std::string projectName = "";
    std::string topicName = "";

    try
    {
        client.CreateTopic(projectName, topicName, shardCount, lifeCycle, type, schema, comment);
    }
    catch (const DatahubException& e)
    {
        std::cerr << "Create topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;

    }
}

創建Blob Topic

 void CreateBlobTopic()
{
    /* Create Topic */
    int shardCount = 3;
    int lifeCycle = 7;
    RecordType type = BLOB;
    std::string projectName = "";
    std::string topicName = "";
    try
    {
        client.CreateTopic(projectName, topicName, shardCount, lifeCycle, type, comment);
    }
    catch (const DatahubException& e)
    {
        std::cerr << "Create topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
}

}

刪除Topic

void DeleteTopic()
{
    std::string projectName = "";
    std::string topicName = "";
    try
    {
       client.DeleteTopic(projectName, topicName);  
    }
    catch(const DatahubException& e)
    {
        std::cerr << "Delete topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }

}

獲取Topic列表

void ListTopic(){
    std::string projectName = "";
    try
    {
        const ListTopicResult& listTopicResult = client.ListTopic(projectName);
        std::cout<<listTopicResult.GetTopicNames().size()<<std::endl;
    }
    catch(const DatahubException& e)
    {
        std::cerr << "Get topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }

}

更新Topic

void UpdateTopic(){
    const std::string updateComment = "test1";
    int updateLifecycle = 7;
    std::string projectName = "";
    std::string topicName = "";
    try
    {
        client.UpdateTopic(projectName, topicName, updateLifecycle, updateProjectComment);
    }
    catch(const DatahubException& e)
    {
        std::cerr << "Update topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }

}

查詢Topic

void GetTopic(){
    std::string projectName = "";
    std::string topicName = "";

    try
    {
        const GetTopicResult& getTopicResult = client.GetTopic(projectName, topicName);
        cout<<getTopicResult.GetComment()<<endl;
    }
    catch(const DatahubException& e)
    {
        std::cerr << "Get topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }

}

Shard操作

Shard表示對一個Topic進行數據傳輸的并發通道，每個Shard會有對應的ID。每個Shard會有多種狀態：Opening - 啟動中，Active - 啟動完成可服務。每個Shard啟用以后會占用一定的服務端資源，建議按需申請Shard數量。

列出Shard列表

void ListShard()
{
    try
    {
       const ListShardResult& lsr = client.ListShard(projectName, topicName);
       std::cout<<lsr.GetShards().size()<<std::endl;
       std::vector<ShardEntry> shards = lsr.GetShards();

       std::vector<std::string> shardIds;
       for (size_t i = 0; i < shards.size(); ++i)
       {		
            ShardEntry shardEntry = shards[i];
            shardIds.push_back(shardEntry.GetShardId());
       }
    } 
    catch(const DatahubException& e)
    {
        std::cerr << "List shard fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
}

分裂Shard

void SplitShard()
{
  std::string projectName = "";
  std::string topicName = "";
  try
  {
    const SplitShardResult& ssr = client.SplitShard(projectName, topicName, "0", "00000000000000000000000000AAAAAA");
    std::cout<<ssr.GetChildShards().size()<<std::endl;
  }
    catch(const DatahubException& e)
    {
        std::cerr << "Split shardId fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
}

合并Shard

void MergeShard()
{
  std::string projectName = "";
  std::string topicName = "";

  try
  {
    const MergeShardResult& msr = client.MergeShard(projectName, topicName, "0", "1");
    std::cout<<msr.GetChildShard().GetShardId()<<std::endl;
    std::cout<<msr.GetChildShard().GetBeginHashKey()<<std::endl;
    std::cout<<msr.GetChildShard().GetEndHashKey()<<std::endl;
  }
   catch(const DatahubException& e)
   {
        std::cerr << "Merge sahrd fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
	
}

讀寫數據

狀態為CLOSED和ACTIVE的shard都可以讀取數據，不過只有狀態為ACTIVE的shard可以寫數據。

讀數據

獲取cursor

讀取Topic下的數據，需要指定對應的shard，同時需要指定數據讀取的游標位置Cursor。Cursor的獲取方式有四種，分別是OLDEST, LATEST, SEQUENCE, SYSTEM_TIME。

OLDEST : 表示獲取的cursor指向當前有效數據中時間最久遠的record。
LATEST : 表示獲取的cursor指向當前最新的record。
SEQUENCE : 表示獲取的cursor指向該序列的record。
SYSTEM_TIME : 表示獲取的cursor指向該大于等于該時間戳的第一條record。

void GetCursor()
{
    try
   {
      const GetCursorResult& r1 = client.GetCursor(projectName, topicName, "0", CURSOR_TYPE_OLDEST);
      std::string cursor = r1.GetCursor();
   } 
    catch(const DatahubException& e)
    {
        std::cerr << "Get cursor fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
	
}

讀取數據

void sub(DatahubClient& client)
{
    std::string projectName = "";
    std::string topicName = "";
    std::string shardId = "0";
    std::string cursor;
    RecordSchema schema;
    OpenSubscriptionOffsetSessionResult osor = client.InitSubscriptionOffsetSession(projectName, topicName, subId, {shardId});
    SubscriptionOffset subscription = osor.GetOffsets().at(shardId);
    if (subscription.GetSequence() < 0)
    {
        cursor = client.GetCursor(projectName, topicName, shardId, CURSOR_TYPE_OLDEST).GetCursor();
    }
    else
    {
        int64_t nextSequence = subscription.GetSequence() + 1;
        try
        {
            cursor = client.GetCursor(projectName, topicName, shardId, CURSOR_TYPE_SEQUENCE, nextSequence).GetCursor();
        }
        catch (const DatahubException& e)
        {
            cursor = client.GetCursor(projectName, topicName, shardId, CURSOR_TYPE_OLDEST).GetCursor();
        }
    }

    int64_t readTotal = 0;
    int fetchNum = 10;
    while (true)
    {
        try
        {
            GetRecordResult grr = client.GetRecord(projectName, topicName, shardId, cursor, fetchNum, subId);
            if (grr.GetRecordCount() <= 0)
            {
                std::cout << "Read null, wait for 1s." << std::endl;
                sleep(1);
                continue;
            }
            
            for (auto recordResult : grr.GetRecords())
            {
                ProcessRecords(recordResult);
                if (++readTotal % 1000 == 0)
                {
                    subscription.SetSequence(recordResult.GetSequence());
                    subscription.SetTimestamp(recordResult.GetSystemTime());
                    std::map<std::string, SubscriptionOffset> offsets;
                    offsets[shardId] = subscription;
                    try
                    {
                        client.UpdateSubscriptionOffset(projectName, topicName, subId, offsets);
                    }
                    catch (const DatahubException& e)
                    {
                        std::cerr << "Update subscription offset fail. requestId: "<< e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
                        throw ;
                    }
                }
            }
            cursor = grr.GetNextCursor();
        }
        catch (const DatahubException& e)
        {
            std::cerr << "Get record fail. requestId: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
        }
    }

inline void ProcessRecords(const RecordResult& record)
{
    if (schema.GetFieldCount() == 0)       // BLOB
    {
        int len = 0;
        const char* data = record.GetData(len);
        printf("%s\n", data);
    }
    else                                   // TUPLE
    {
        for (int j = 0; j < schema.GetFieldCount(); ++j)
        {
            const Field &field = schema.GetField(j);
            const FieldType fieldType = field.GetFieldType();
            switch (fieldType)
            {
                case BIGINT:
                    printf("%ld\n", record.GetBigint(j));
                    break;
                case DOUBLE:
                    printf("%.15lf\n", record.GetDouble(j));
                    break;
                case STRING:
                    printf("%s\n", record.GetString(j).c_str());
                    break;
                default:
                    break;
            }
        }
    }
} 
}

協同消費數據示例：

using namespace aliyun;
using namespace aliyun::datahub;

std::atomic_int gTotalNum(0);

Account gAccount;
std::string gEndpoint;
std::string gProjectName;
std::string gTopicName;
std::string gSubId;
bool gAutoAck;
int32_t gFetchNum;
int64_t gSessionTimeoutMs;

void Usage(const char* prog)
{
    const char *start = strrchr(prog, '/');
    start = (start != NULL) ? start + 1 : prog;
    fprintf(stdout,
        "%s Ver %s\n"
        "Usage: <programme> [options...]\n"
        "   -i --accessid           <string>          set access id [Mandatory]\n"
        "   -k --accesskey          <string>          set access key [Mandatory]\n"
        "   -e --endpoint           <string>          set endpoint [Mandatory]\n"
        "   -p --project            <string>          set project [Mandatory]\n"
        "   -t --topic              <string>          set topic [Mandatory]\n"
        "   -s --subId              <string>          set subscription id [Mandatory]\n"
        "   -a --autoAck            <bool>            set auto ack offset\n"
        "   -f --fetchNum           <int>             set fetch limit num\n"
        "   -S --sessionTimeoutMs   <int>             set session timeout (ms)\n"
        "   -v --version                              show version\n"
        "   -h --help                                 show help message\n", start, PROG_VERSION);
}

void parse_args(int argc, char* argv[])
{
    int opt_id;
    static struct option long_opts[] =
    {
        { "accessid", 1, 0, 'i'},
        { "accesskey", 1, 0, 'k'},
        { "endpoint", 1, 0, 'e' },
        { "project", 1, 0, 'p' },
        { "topic", 1, 0, 't' },
        { "subId", 1, 0, 's' },
        { "autoAck", 1, 0, 'a'},
        { "fetchNum", 1, 0, 'f'},
        { "sessionTimeoutMs", 1, 0, 'S'},
        { "version", 0, 0, 'v' },
        { "help", 0, 0, 'h' },
        { NULL, 0, 0, 0 }
    };

    while (1)
    {
        int ret = getopt_long(argc, argv, "i:k:e:p:t:s:a:f:S:vh", long_opts, &opt_id);
        if (ret == -1)
        {
            break;
        }
        switch (ret)
        {
        case 'i':
            gAccount.id = optarg;
            break;
        case 'k':
            gAccount.key = optarg;
            break;
        case 'e':
            gEndpoint = optarg;
            break;
        case 'p':
            gProjectName = optarg;
            break;
        case 't':
            gTopicName = optarg;
            break;
        case 's':
            gSubId = optarg;
            break;
        case 'a':
            gAutoAck = (bool)std::atoi(optarg);
            break;
        case 'f':
            gFetchNum = std::atoi(optarg);
            break;
        case 'S':
            gSessionTimeoutMs = (int64_t)std::atoi(optarg);
            break;
        case 'v':
        case 'h':
            Usage(argv[0]);
            exit(0);
            break;
        default:
            break;
        }
    }
}

inline void ProcessRecords(const TopicMetaPtr& topicMeta, const RecordResultPtr& recordPtr)
{
    const RecordSchema& schema = topicMeta->GetRecordSchema();
    if (schema.GetFieldCount() == 0)        // BLOB
    {
        int len = 0;
        const char* data = recordPtr->GetData(len);
        printf("%s\n", data);
    }
    else                                    // TUPLE
    {
        for (int j = 0; j < schema.GetFieldCount(); ++j)
        {
            const Field& field = schema.GetField(j);
            const FieldType fieldType = field.GetFieldType();
            switch (fieldType)
            {
                case BIGINT:
                    printf("%ld ", recordPtr->GetBigint(j));
                    break;
                case INTEGER:
                    printf("%d ", recordPtr->GetInteger(j));
                    break;
                case SMALLINT:
                    printf("%hd ", recordPtr->GetSmallint(j));
                    break;
                case TINYINT:
                    printf("%hhd ", recordPtr->GetTinyint(j));
                    break;
                case DOUBLE:
                    printf("%.5lf ", recordPtr->GetDouble(j));
                    break;
                case FLOAT:
                    printf("%.5f ", recordPtr->GetFloat(j));
                    break;
                case STRING:
                    printf("%s ", recordPtr->GetString(j).c_str());
                    break;
                case BOOLEAN:
                    printf("%s ", recordPtr->GetBoolean(j) ? "true" : "false");
                    break;
                case TIMESTAMP:
                    printf("%ld ", recordPtr->GetTimestamp(j));
                    break;
                case DECIMAL:
                    printf("%s ", recordPtr->GetDecimal(j).c_str());
                    break;
                default:
                    break;
            }
        }
        printf("\n");
    }
}

void CollaborativeConsume(const std::string& project, const std::string& topic, const std::string& subId, const ConsumerConfiguration& conf)
{
    DatahubConsumer consumer(project, topic, subId, conf);
    const TopicMetaPtr& topicMeta = consumer.GetTopicMeta();

    uint64_t readRecordNum = 0;
    try{
        while (true)
        {
            auto recordPtr = consumer.Read(60000);
            if (recordPtr == nullptr)
            {
                break;
            }
            // TODO: 處理數據
            ProcessRecords(topicMeta, recordPtr);
            if (!gAutoAck)
            {
                recordPtr->GetMessageKey()->Ack();      // 如果auto_ack設置為false,則處理完數據執行Ack;
            }
            readRecordNum++;
        }
    }
    catch (const std::exception& e)
    {
        std::cerr << "Read fail: " << e.what() << std::endl;
    }
    std::cout << "Read " << readRecordNum << " records total" << std::endl;
}

int main(int argc, char* argv[])
{
    parse_args(argc, argv);

    ConsumerConfiguration consumerConf(gAccount, gEndpoint);
    consumerConf.SetLogFilePath("./DatahubCollaborativeConsumer.log");
    consumerConf.SetAutoAckOffset(gAutoAck);
    if (gFetchNum > 0)
    {
        consumerConf.SetFetchLimitNum(gFetchNum);
    }
    if (gSessionTimeoutMs > 0)
    {
        consumerConf.SetSessionTimeout(gSessionTimeoutMs);
    }

    CollaborativeConsume(gProjectName, gTopicName, gSubId, consumerConf);

    return 0;
}

說明

協同消費為sdk 2.25版本新增功能，之前版本不支持

寫數據

void pub(DatahubClient& client)
{
    std::vector<RecordEntry> records;
    GenerateRecords(records);

    try
    {
        PutRecordResult prr = client.PutRecord(projectName, topicName, records);
        if (prr.GetFailedRecordCount() > 0)
        {
            pubRetry(client, prr.GetFailedRecords(), 3);
        }
    }
    catch (const DatahubException& e)
    {
        std::cerr << "Put records fail. requestId: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
}

void pubRetry(DatahubClient& client, const std::vector<RecordEntry>& records, int retryTimes)
{
    if (retryTimes <= 0)
    {
        std::cout << "Put record retry fail. records size: " << records.size() << std::endl;
        return ;
    }

    try
    {
        PutRecordResult prr = client.PutRecord(projectName, topicName, records);
        if (prr.GetFailedRecordCount() > 0)
        {
            pubRetry(client, prr.GetFailedRecords(), retryTimes-1);
        }
        else
        {
            std::cout << "Put records retry success." << std::endl;
        }
    }
    catch (const DatahubException& e)
    {
        std::cerr << "Put records fail when put retry. requestId: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
    }
inline void GenerateRecords(RecordEntryVec& records)
{
    for (int i = 0; i < 10; i++)
    {   
        if (schema.GetFieldCount() == 0) 
        {        
             // BLOB
            RecordEntry record = RecordEntry(BLOB);
            record.SetData("test_blob_data" + std::to_string(i));
            records.push_back(record);
        }
        else
        {
        // TUPLE
            // RecordEntry record = RecordEntry(5);
            // record.SetString(0, "field_1_" + std::to_string(i));
            // record.SetBigint(1, 123456789l);
            // record.SetDouble(2, 123.456d);
            // record.SetDouble(3, 654.32100d);
            // record.SetString(4, "field_2_" + std::to_string(i));
            // records.push_back(record);
        } 

    }
}  
}

異步寫入示例:

using namespace aliyun;
using namespace aliyun::datahub;

Account gAccount;
std::string gEndpoint;
std::string gProjectName;
std::string gTopicName;
int gEpochNum = 10;
int gRecordNum = 10;
int64_t gMaxBufferRecords = 0l;
int64_t gMaxBufferSize = 0l;
int64_t gMaxBufferTimeMs = 0l;
int64_t gMaxRecordPackQueueLimit = 0l;
StringVec gShardIds;

inline void StringSplit(StringVec& tar, const std::string& src, const std::string& pre)
{
    size_t start_pos = 0, end_pos = src.size();
    size_t pos = src.find(pre, start_pos);
    while (pos != std::string::npos)
    {
        std::string addStr = src.substr(start_pos, pos - start_pos);
        if (!addStr.empty())
        {
            tar.push_back(addStr);
        }
        start_pos = pos + 1;
        pos = src.find(pre, start_pos);
    }
    std::string addStr = src.substr(start_pos, end_pos - start_pos);
    if (!addStr.empty())
    {
        tar.push_back(addStr);
    }
}

void Usage(const char* prog)
{
    const char *start = strrchr(prog, '/');
    start = (start != NULL) ? start + 1 : prog;
    fprintf(stdout,
        "%s Ver %s\n"
        "Usage: <programme> [options...]\n"
        "   -i --accessid           <string>          set access id [Mandatory]\n"
        "   -k --accesskey          <string>          set access key [Mandatory]\n"
        "   -e --endpoint           <string>          set endpoint [Mandatory]\n"
        "   -p --project            <string>          set project [Mandatory]\n"
        "   -t --topic              <string>          set topic [Mandatory]\n"
        "   -E --epochNum           <int>             set epoch num for write [Mandatory]\n"
        "   -N --recordNum          <int>             set record num for each epoch [Mandatory]\n"
        "   -R --maxBufferRecords   <int>             set max buffer record count\n"
        "   -S --maxBufferSize      <int>             set max buffer size\n"
        "   -M --maxBufferTimeMs    <int>             set max buffer time (ms)\n"
        "   -Q --maxRecordPackLimit <int>             set max record pack queue limit\n"
        "   -H --shardIds           <string>          set shards (split by ',')\n"
        "   -v --version                              show version\n"
        "   -h --help                                 show help message\n", start, PROG_VERSION);
}

void parse_args(int argc, char* argv[])
{
    int opt_id;
    static struct option long_opts[] =
    {
        { "accessid", 1, 0, 'i'},
        { "accesskey", 1, 0, 'k'},
        { "endpoint", 1, 0, 'e' },
        { "project", 1, 0, 'p' },
        { "topic", 1, 0, 't' },
        { "epochNum", 1, 0, 'E'},
        { "recordNum", 1, 0, 'N'},
        { "maxBufferRecords", 1, 0, 'R'},
        { "mMaxBufferSize", 1, 0, 'S'},
        { "maxBufferTimeMs", 1, 0, 'M'},
        { "maxRecordPackQueueLimit", 1, 0, 'Q'},
        { "shardIds", 1, 0, 'H'},
        { "version", 0, 0, 'v' },
        { "help", 0, 0, 'h' },
        { NULL, 0, 0, 0 }
    };

    while (1)
    {
        int ret = getopt_long(argc, argv, "i:k:e:p:t:E:N:R:S:M:Q:H:vh", long_opts, &opt_id);
        if (ret == -1)
        {
            break;
        }
        switch (ret)
        {
        case 'i':
            gAccount.id = optarg;
            break;
        case 'k':
            gAccount.key = optarg;
            break;
        case 'e':
            gEndpoint = optarg;
            break;
        case 'p':
            gProjectName = optarg;
            break;
        case 't':
            gTopicName = optarg;
            break;
        case 'E':
            gEpochNum = std::atoi(optarg);
            break;
        case 'N':
            gRecordNum = std::atoi(optarg);
            break;
        case 'R':
            gMaxBufferRecords = (int64_t)std::atoi(optarg);
            break;
        case 'S':
            gMaxBufferSize = (int64_t)std::atoi(optarg);
            break;
        case 'M':
            gMaxBufferTimeMs = (int64_t)std::atoi(optarg);
            break;
        case 'Q':
            gMaxRecordPackQueueLimit = (int64_t)std::atoi(optarg);
            break;
        case 'H':
            StringSplit(gShardIds, optarg, ",");
            break;
        case 'v':
        case 'h':
            Usage(argv[0]);
            exit(0);
            break;
        default:
            break;
        }
    }

    if (gEpochNum <= 0 || gRecordNum <= 0)
    {
        std::cerr << "Invalid parameter!" << std::endl;
        exit(1);
    }
}

inline void GenerateTupleRecord(const RecordSchema& schema, RecordEntry& record)
{
    for (int i = 0; i < schema.GetFieldCount(); i++)
    {
        const Field& field = schema.GetField(i);
        const FieldType& type = field.GetFieldType();
        switch (type)
        {
            case BIGINT:
                record.SetBigint(i, 1234l);
                break;
            case DOUBLE:
                record.SetDouble(i, 1.234);
                break;
            case BOOLEAN:
                record.SetBoolean(i, true);
                break;
            case TIMESTAMP:
                record.SetTimestamp(i, 1234l);
                break;
            case STRING:
                record.SetString(i, "1234");
                break;
            case DECIMAL:
                record.SetDecimal(i, "1234");
                break;
            case INTEGER:
                record.SetInteger(i, (int32_t)1234);
                break;
            case FLOAT:
                record.SetFloat(i, 1.234);
                break;
            case TINYINT:
                record.SetTinyint(i, (int8_t)1234);
                break;
            case SMALLINT:
                record.SetSmallint(i, (int16_t)1234);
                break;
            default:
                break;
        }
    }
}

inline void GenerateRecords(TopicMetaPtr topicMeta, RecordEntryVec& records)
{
    const RecordSchema& schema = topicMeta->GetRecordSchema();
    for (int i = 0; i < gRecordNum; i++)
    {
        if (topicMeta->GetRecordType() == "TUPLE")
        {
            // TUPLE
            records.emplace_back(schema.GetFieldCount());
            GenerateTupleRecord(schema, records.back());
        }
        else
        {
            // BLOB
            records.emplace_back(BLOB);
            records.back().SetData("test_blob_data" + std::to_string(i));
        }
    }
}

void AsyncProduce(const std::string& project, const std::string& topic, const ProducerConfiguration& conf, const StringVec& shardIds)
{
    DatahubProducer producer(project, topic, conf, shardIds);
    const TopicMetaPtr& topicMeta = producer.GetTopicMeta();

    std::vector<WriteResultFuturePtr> resultFutureVecs;
    std::map<std::string, uint64_t> writeRecordNum;
    for (const std::string& shardId : shardIds)
    {
        writeRecordNum[shardId] = 0;
    }

    try
    {
        // TODO: 生成數據
        RecordEntryVec records;
        GenerateRecords(topicMeta, records);
        for (int i = 0; i < gEpochNum; i++)
        {
            auto result = producer.WriteAsync(records);
            resultFutureVecs.push_back(result);
        }
    }
    catch(const std::exception& e)
    {
        std::cerr << "WriteAsync fail: " << e.what() << std::endl;
    }

    producer.Flush();

    for (auto it = resultFutureVecs.begin(); it != resultFutureVecs.end(); it++)
    {
        (*it)->wait();
        try
        {
            WriteResultPtr result = (*it)->get();
            writeRecordNum[result->GetShardId()] += gRecordNum;
        }
        catch (const std::exception& e)
        {
            std::cerr << "Write records fail: " << e.what() << std::endl;
        }
    }

    for (auto it = writeRecordNum.begin(); it != writeRecordNum.end(); it++)
    {
        std::cout << "Write " << it->second << " records to shard " << it->first << std::endl;
    }
}

int main(int argc, char *argv[])
{
    parse_args(argc, argv);

    ProducerConfiguration producerConf(gAccount, gEndpoint);
    producerConf.SetEnableProtobuf(true);
    producerConf.SetLogFilePath("./DatahubAsyncProducer.log");
    if (gMaxBufferRecords > 0)
    {
        producerConf.SetMaxAsyncBufferRecords(gMaxBufferRecords);
    }
    if (gMaxBufferSize > 0)
    {
        producerConf.SetMaxAsyncBufferSize(gMaxBufferSize);
    }
    if (gMaxBufferTimeMs > 0)
    {
        producerConf.SetMaxAsyncBufferTimeMs(gMaxBufferTimeMs);
    }
    if (gMaxRecordPackQueueLimit > 0)
    {
        producerConf.SetMaxRecordPackQueueLimit(gMaxRecordPackQueueLimit);
    }

    AsyncProduce(gProjectName, gTopicName, producerConf, gShardIds);

    return 0;
}

同步寫入示例

using namespace aliyun;
using namespace aliyun::datahub;

Account gAccount;
std::string gEndpoint;
std::string gProjectName;
std::string gTopicName;
int gEpochNum = 10;
int gRecordNum = 10;
StringVec gShardIds;

inline void StringSplit(StringVec& tar, const std::string& src, const std::string& pre)
{
    size_t start_pos = 0, end_pos = src.size();
    size_t pos = src.find(pre, start_pos);
    while (pos != std::string::npos)
    {
        std::string addStr = src.substr(start_pos, pos - start_pos);
        if (!addStr.empty())
        {
            tar.push_back(addStr);
        }
        start_pos = pos + 1;
        pos = src.find(pre, start_pos);
    }
    std::string addStr = src.substr(start_pos, end_pos - start_pos);
    if (!addStr.empty())
    {
        tar.push_back(addStr);
    }
}

void Usage(const char* prog)
{
    const char *start = strrchr(prog, '/');
    start = (start != NULL) ? start + 1 : prog;
    fprintf(stdout,
        "%s Ver %s\n"
        "Usage: <programme> [options...]\n"
        "   -i --accessid           <string>          set access id [Mandatory]\n"
        "   -k --accesskey          <string>          set access key [Mandatory]\n"
        "   -e --endpoint           <string>          set endpoint [Mandatory]\n"
        "   -p --project            <string>          set project [Mandatory]\n"
        "   -t --topic              <string>          set topic [Mandatory]\n"
        "   -E --epochNum           <int>             set epoch num for write [Mandatory]\n"
        "   -N --recordNum          <int>             set record num for each epoch [Mandatory]\n"
        "   -H --shardIds           <string>          set shards (split by ',')\n"
        "   -v --version                              show version\n"
        "   -h --help                                 show help message\n", start, PROG_VERSION);
}

void parse_args(int argc, char* argv[])
{
    int opt_id;
    static struct option long_opts[] =
    {
        { "accessid", 1, 0, 'i'},
        { "accesskey", 1, 0, 'k'},
        { "endpoint", 1, 0, 'e' },
        { "project", 1, 0, 'p' },
        { "topic", 1, 0, 't' },
        { "epochNum", 1, 0, 'E'},
        { "recordNum", 1, 0, 'N'},
        { "shardIds", 1, 0, 'H'},
        { "version", 0, 0, 'v' },
        { "help", 0, 0, 'h' },
        { NULL, 0, 0, 0 }
    };

    while (1)
    {
        int ret = getopt_long(argc, argv, "i:k:e:p:t:E:N:H:vh", long_opts, &opt_id);
        if (ret == -1)
        {
            break;
        }
        switch (ret)
        {
        case 'i':
            gAccount.id = optarg;
            break;
        case 'k':
            gAccount.key = optarg;
            break;
        case 'e':
            gEndpoint = optarg;
            break;
        case 'p':
            gProjectName = optarg;
            break;
        case 't':
            gTopicName = optarg;
            break;
        case 'E':
            gEpochNum = std::atoi(optarg);
            break;
        case 'N':
            gRecordNum = std::atoi(optarg);
            break;
        case 'H':
            StringSplit(gShardIds, optarg, ",");
            break;
        case 'v':
        case 'h':
            Usage(argv[0]);
            exit(0);
            break;
        default:
            break;
        }
    }

    if (gEpochNum <= 0 || gRecordNum <= 0)
    {
        std::cerr << "Invalid parameter!" << std::endl;
        exit(1);
    }
}

inline void GenerateTupleRecord(const RecordSchema& schema, RecordEntry& record)
{
    for (int i = 0; i < schema.GetFieldCount(); i++)
    {
        const Field& field = schema.GetField(i);
        const FieldType& type = field.GetFieldType();
        switch (type)
        {
            case BIGINT:
                record.SetBigint(i, 1234l);
                break;
            case DOUBLE:
                record.SetDouble(i, 1.234);
                break;
            case BOOLEAN:
                record.SetBoolean(i, true);
                break;
            case TIMESTAMP:
                record.SetTimestamp(i, 1234l);
                break;
            case STRING:
                record.SetString(i, "1234");
                break;
            case DECIMAL:
                record.SetDecimal(i, "1234");
                break;
            case INTEGER:
                record.SetInteger(i, (int32_t)1234);
                break;
            case FLOAT:
                record.SetFloat(i, 1.234);
                break;
            case TINYINT:
                record.SetTinyint(i, (int8_t)1234);
                break;
            case SMALLINT:
                record.SetSmallint(i, (int16_t)1234);
                break;
            default:
                break;
        }
    }
}

inline void GenerateRecords(const TopicMetaPtr& topicMeta, RecordEntryVec& records)
{
    const RecordSchema& schema = topicMeta->GetRecordSchema();
    for (int i = 0; i < gRecordNum; i++)
    {
        if (topicMeta->GetRecordType() == "TUPLE")
        {
            // TUPLE
            records.emplace_back(schema.GetFieldCount());
            GenerateTupleRecord(schema, records.back());
        }
        else
        {
            // BLOB
            records.emplace_back(BLOB);
            records.back().SetData("test_blob_data" + std::to_string(i));
        }
    }
}

void GeneralProduce(const std::string& project, const std::string& topic, const ProducerConfiguration& conf, const StringVec& shardIds)
{
    DatahubProducer producer(project, topic, conf, shardIds);
    const TopicMetaPtr& topicMeta = producer.GetTopicMeta();

    std::map<std::string, uint64_t> writeRecordNum;
    for (const std::string& shardId : shardIds)
    {
        writeRecordNum[shardId] = 0;
    }

    try
    {
        // TODO: 生成數據
        RecordEntryVec records;
        GenerateRecords(topicMeta, records);
        for (int i = 0; i < gEpochNum; i++)
        {
            std::string shardId = producer.Write(records);
            writeRecordNum[shardId] += records.size();
        }
    }
    catch(const std::exception& e)
    {
        std::cerr << "Write fail: " << e.what() << std::endl;
    }

    for (auto it = writeRecordNum.begin(); it != writeRecordNum.end(); it++)
    {
        std::cout << "Write " << it->second << " records to shard " << it->first << std::endl;
    }
}

int main(int argc, char *argv[])
{
    parse_args(argc, argv);

    ProducerConfiguration producerConf(gAccount, gEndpoint);
    producerConf.SetEnableProtobuf(true);
    producerConf.SetLogFilePath("./DatahubGeneralProducer.log");

    GeneralProduce(gProjectName, gTopicName, producerConf, gShardIds);

    return 0;
}

Meter操作

void GetMeter()
{
    try
    {
        GetMeteringInfoResult GetMeteringInfo = Client.GetMeteringInfo(projectName, topicName, shardId);

    }
    catch(DatahubException e)
    {
          cerr << e << endl;
    }
}

說明

異步寫入、同步寫入從sdk 2.25版本起支持

Subscribtion操作

訂閱服務提供了服務端保存用戶消費點位的功能，只需要通過簡單配置和處理，就可以實現高可用的點位存儲服務。

創建 Subscription

void CreateSubscription()
{ 
  std::string projectName = "";
  std::string topicName = "";
  try
  {
    const std::string subscriptionComment = "test_subscription";
    const CreateSubscriptionResult& createSubscriptionResult = client.CreateSubscription(projectName, topicName, subscriptionComment);
  } 
   catch(const DatahubException& e)
   {
        std::cerr << "Create sub fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

刪除 Subscription

void DeleteSubscription() 
{
  std::string projectName = "";
  std::string topicName = "";
  std::string subId = "";
  try 
  {
     client.DeleteSubscription(projectName, topicName, subId);
  } 
   catch(const DatahubException& e)
   {
      std::cerr << "Delete sub fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

更新 Subscription

void UpdateSubscription()
{
    try 
    {
        const std::string updateSubscriptionComment = "test_subscription_1";
      	const std::string projectName = "";
        const std::string topicName = "";
        client.UpdateSubscription(projectName, topicName, subId, updateSubscriptionComment);
      }
   catch(const DatahubException& e)
   {
      std::cerr << "Upate sub fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

獲取Subscription列表

 void ListSubscription() 
 {
    const std::string projectName = "";
    const std::string topicName = "";
    const std::string subId = "";

    try 
    {
       const ListSubscriptionResult& subscriptionResult = client.ListSubscription(projectName, topicName, 1, 20, subId));
       std::cout << subscriptionResult.GetTotalCount() << std::endl;
    }  
   catch(const DatahubException& e)
   {
      std::cerr << "list sub fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

查詢 Subscription

void GetSubscription() 
{
    try 
    {
    	const std::vector<SubscriptionEntry>& subscriptions = subscriptionResult.GetSubscriptions();
    	std::cout << subscriptions.size() << std::endl;

    	for (size_t i = 0; i < subscriptions.size(); ++i)
    	{
          SubscriptionEntry entry = subscriptions[i];
          std::cout <<  entry.GetSubId() << std::endl;
      }						
    }  
   catch(const DatahubException& e)
   {
      std::cerr << "Get sub fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

更新 Subscription 狀態

 void UpdateSubscriptionState()
 {
    try 
    {
        client.UpdateSubscriptionState(projectName, topicName, subId, SubscriptionState::OFFLINE);
        std::cout << getSubscriptionResult.GetComment() << std::endl;

    } 
   catch(const DatahubException& e)
   {
      std::cerr << "Update sub fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
	
}

offset 操作

一個subscription創建后，初始狀態是未消費的，要使用subscription服務提供的點位存儲功能，需要進行一些offset操作。

初始化offset

 void InitSubscriptionOffsetSession()
 {
    const std::string projectName = "";
    const std::string topicName = "";
    const std::string subId = "";
    std::vector<std::string> shardIds;
    try 
    {
        const OpenSubscriptionOffsetSessionResult& offsetSessionResult =
        client.InitSubscriptionOffsetSession(projectName, topicName, subId, shardIds);
        std::cout << offsetSessionResult.GetOffsets().size() << std::endl;

    } 
   catch(const DatahubException& e)
   {
      std::cerr << "Init offset fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

獲取 offset

//獲取點位
void GetSubscriptionOffset() 
{
    const std::string projectName = "";
    const std::string topicName = "";
    const std::string subId = "";
    std::vector<std::string> shardIds;
    try 
    {
        const GetSubscriptionOffsetResult& getSubscriptionOffsetResult =
        client.GetSubscriptionOffset(projectName, topicName, subId, shardIds;
        std::cout << getSubscriptionOffsetResult.GetOffsets().size() << std::endl;        std::cout << getSubscriptionOffsetResult.GetOffsets().size() << std::endl;
        std::cout << getSubscriptionResult.GetComment() << std::endl;

    }  
   catch(const DatahubException& e)
   {
      std::cerr << "Get offset fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

重置 offset

void ResetSubscriptionOffset() 
{
    try 
    {
        int64_t resetTimestamp = 0l;
        int64_t resetSequence = 0l;
        uint32_t resetBatchIndex = 0u;
        const std::string projectName = "";
        const std::string topicName = "";
        const std::string subId = "";
        std::map<std::string, SubscriptionOffset> resetSubscriptionOffsets;
        for (auto iter = offsets.begin(); iter != offsets.end(); ++iter)
        {
            SubscriptionOffset offset(resetTimestamp, resetSequence, resetBatchIndex);
            resetSubscriptionOffsets.insert(
            std::pair<std::string, SubscriptionOffset>(iter->first, offset));
        }

    client.ResetSubscriptionOffset(projectName, topicName, subId, resetSubscriptionOffsets);    }  
   catch(const DatahubException& e)
   {
      std::cerr << "Reset offset fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

Connector 操作

DataHub Connector是把DataHub服務中的流式數據同步到其他云產品中的功能，目前支持將Topic中的數據實時/準實時同步到MaxCompute(ODPS)、Oss、RDS&Mysql、TableStore、Oss、ElasticSearch、函數計算中。用戶只需要向DataHub中寫入一次數據，并在DataHub服務中配置好同步功能，便可以在其他云產品中使用這份數據。

創建 Connector

ODPS示例

void  CreateConnector() 
{
    try{
    sdk::SinkOdpsConfig config;
    config.SetEndpoint("ODPS_ENDPOINT");
    config.SetProject("ODPS_PROJECT");
    config.SetTable("ODPS_TABLE");
    config.SetAccessId("ODPS_ACCESSID");
    config.SetAccessKey("ODPS_ACCESSKEY");
    config.SetPartitionMode(sdk::PartitionMode::SYSTEM_TIME);
    config.SetTimeRange(15);
    const std::string projectName = "";
    const std::string topicName = "";
    std::vector<std::pair<std::string, std::string> > partitionConfig;
    partitionConfig.push_back(std::pair<std::string, std::string>("ds", "%Y%m%d"));
    partitionConfig.push_back(std::pair<std::string, std::string>("hh", "%H"));
    partitionConfig.push_back(std::pair<std::string, std::string>("mm", "%M"));
    config.SetPartitionConfig(partitionConfig);

    std::vector<std::string> columnFields;
    columnFields.push_back(fieldName1);

    const CreateConnectorResult& connectorResult = client.CreateConnector(projectName, topicName, sdk::ConnectorType::SINK_ODPS, columnFields, config);

    std::cout<<"Odps ConnectorId:" + connectorResult.GetConnectorId()<<std::endl;
    }
   catch(const DatahubException& e)
   {
      std::cerr << "Create odpsConnector fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

刪除 Connector

void DeleteConnector() 
{
    const std::string projectName = "";
    const std::string topicName = "";
    const std::string connectorId = "";

    try {
      client.DeleteConnector(projectName, topicName, connectorId);
      } 
   	catch(const DatahubException& e)
   {
      std::cerr << "Delete Connector fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

查詢 Connector

void GetConnector() 
{
    const std::string projectName = "";
    const std::string topicName = "";
    const std::string connectorId = "";
    try 
    {
        GetConnectorResult GetConnectorResult = client.GetConnector(projectName, topicName, connectorId);
        const sdk::SinkOdpsConfig* odpsConfig = dynamic_cast<const sdk::SinkOdpsConfig*>(getConnectorResult.GetConfig());
        std::cout << odpsConfig->GetPartitionConfig().size() << std::endl;
    } 
   catch(const DatahubException& e)
   {
      std::cerr << "Delete topic fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
}

列出 Connector列表

void ListConnector() {
  try {
       const std::string projectName = "";
       const std::string topicName = "";
       const ListConnectorResult& listConnectorResult = client.ListConnector(projectName, topicName);
       std::cout << listConnectorResult.GetConnectorIds().size() << std::endl;
  }  catch(const DatahubException& e)
   {
      std::cerr << "list connector fail: " << e.GetRequestId() << ", ErrorCode: " << e.GetErrorCode() << ", ErrorMessage: " << e.GetErrorMessage() << std::endl;
   }
 }

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

安裝

初始化