SDK介紹__使用指南_DataHub實時數據通道_大數據計算服務-阿裏雲

更多寫入方式請參考多方式數據寫入

消費Tuple數據

獲取Cursor，可以通過三種方式獲取：”OLDEST”, “LATEST”, “SYSTEM_TIME”

GetCursorResult cursorRs = 
    client.getCursor(projectName, topicName, shardId, GetCursorRequest.CursorType.OLDEST);
//GetCursorResult cursorRs = client.getCursor(projectName, topicName, shardId, System.currentTimeMillis() - 24 * 3600 * 1000 /* ms */); 可以獲取到24小時內的第一條數據Cursor

讀取數據:

int limit = 100;
String cursor = cursorRs.getCursor();
while (true) {
    try {
        GetRecordsResult recordRs = client.getRecords(projectName, topicName, shardId, cursor, limit, schema);
        List<RecordEntry> recordEntries = recordRs.getRecords();
        if (cursor.equals(recordRs.getNextCursor())) {
            // no more data , please wait
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        cursor = recordRs.getNextCursor();
    } catch (InvalidCursorException ex) {
        cursorRs = client.getCursor(projectName, topicName, shardId, GetCursorRequest.CursorType.OLDEST);
        cursor = cursorRs.getCursor();
    }
}

BlobTopic寫入二進製數據使用樣例

創建Blob Topic

int shardCount = 5;
int lifeCycle = 3;
String topicName = "topic_example";
String topicDesc = "topic_example_desc";
client.createTopic(projectName, topicName, shardCount, lifeCycle, RecordType.BLOB, topicDesc);

List<BlobRecordEntry> recordEntries = new ArrayList<BlobRecordEntry>();
String shardId = listShardResult.getShards().get(0).getShardId();
String data = String.valueOf(System.currentTimeMillis());
BlobRecordEntry entry = new BlobRecordEntry();
entry.setData(data.getBytes());
entry.setShardId(shardId);
recordEntries.add(entry);
PutBlobRecordsResult result = client.putBlobRecords(projectName, topicName, recordEntries);
if (result.getFailedRecordCount() != 0) {
    List<ErrorEntry> errors = result.getFailedRecordError();
    // deal with result.getFailedRecords()
}

int limit = 100;
String cursor = cursorRs.getCursor();
while (true) {
    try {
        GetBlobRecordsResult recordRs = client.getBlobRecords(projectName, topicName, shardId, cursor, limit);
        List<BlobRecordEntry> recordEntries = recordRs.getRecords();
        Assert.assertEquals(recordRs.getRecords().get(0).getData(), data.getBytes());
        if (cursor.equals(recordRs.getNextCursor())) {
            // no more data , please wait
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        cursor = recordRs.getNextCursor();
    } catch (InvalidCursorException ex) {
        cursorRs = client.getCursor(projectName, topicName, shardId, GetCursorRequest.CursorType.OLDEST);
        cursor = cursorRs.getCursor();
    }
}