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();
    }
}