表格存储新手指南:Java SDK异步接口的使用-阿里云开发者社区

开发者社区> 阿里云存储服务> 正文
登录阅读全文

表格存储新手指南:Java SDK异步接口的使用

简介: 本篇文章主要会介绍下表格存储的Java SDK提供的异步接口,如何使用以及应用场景。

本篇文章主要会介绍下表格存储的Java SDK提供的异步接口,如何使用以及应用场景。


为什么需要异步?

异步提供了一个non-blocking, event-driven的编程模型,能够将系统不同层级的模块进行层次化的解耦,能够利用多核并行执行任务,提高性能。

现如今,一个大型的系统,系统级调优的最关键一步,就是异步化。异步化最常改造的是远程RPC或者数据库访问部分,表格存储作为一个底层数据库产品,需要提供异步接口来适应这个潮流。


在表格存储内部,我们也有一些使用异步来优化系统的例子,就拿Java SDK来说,可以看下以下两篇文章:

1. 使用NIO来优化Java SDK的性能

2. 基于Java SDK异步接口,提供高并发、高吞吐率的数据导入接口



如何使用?

异步接口的使用和同步接口没有太大区别,使用同样的请求参数,唯一的不同在于返回结果的处理上。同步接口会同步的返回调用结果,而异步接口会返回Future类型的结果,或者直接通过Callback来通知结果。

Future的使用

    private static void listTableWithFuture(OTSClientAsync client) {
        // 通过Future同步的等待结果返回。
        try {
            OTSFuture<ListTableResult> future = client.listTable();
            ListTableResult result = future.get(); // 同步的等待
            System.out.println("\nList table by listTableWithFuture:");
            for (String tableName : result.getTableNames()) {
                System.out.println(tableName);
            }
        } catch (OTSException e) {
            e.printStackTrace();
        } catch (ClientException e) {
            e.printStackTrace();
        }

        // 通过Future,间歇性的等待结果返回。
        try {
            OTSFuture<ListTableResult> future = client.listTable();

            while (!future.isDone()) {
                System.out.println("Waiting for result of list table.");
                Thread.sleep(10); // 每隔10ms检查结果是否返回
            }

            ListTableResult result = future.get();
            System.out.println("\nList table by listTableWithFuture:");
            for (String tableName : result.getTableNames()) {
                System.out.println(tableName);
            }
        } catch (OTSException e) {
            e.printStackTrace();
        } catch (ClientException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

Callback的使用

    private static void listTableWithCallback(OTSClientAsync asyncClient) {
        final AtomicBoolean isDone = new AtomicBoolean(false);
        OTSCallback<ListTableRequest, ListTableResult> callback = new OTSCallback<ListTableRequest, ListTableResult>() {
            @Override
            public void onCompleted(OTSContext<ListTableRequest, ListTableResult> otsContext) {
                isDone.set(true);
                System.out.println("\nList table by listTableWithCallback:");
                for (String tableName : otsContext.getOTSResult().getTableNames()) {
                    System.out.println(tableName);
                }
            }

            @Override
            public void onFailed(OTSContext<ListTableRequest, ListTableResult> otsContext, OTSException ex) {
                isDone.set(true);
                ex.printStackTrace();
            }

            @Override
            public void onFailed(OTSContext<ListTableRequest, ListTableResult> otsContext, ClientException ex) {
                isDone.set(true);
                ex.printStackTrace();
            }
        };

        asyncClient.listTable(callback); // 将callback扔给SDK,SDK在完成请求接到响应后,会自动调用callback

        // 等待callback被调用,一般的业务处理逻辑下,不需要这一步等待。
        while (!isDone.get()) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

案例一:如何突破BatchWriteRow的行数限制,一次性导入N行数据

    private static void batchWriteRow(OTSClientAsync asyncClient, String tableName) {
        // BatchWriteRow的行数限制是100行,使用异步接口,实现一次批量导入1000行。
        List<OTSFuture<BatchWriteRowResult>> futures = new ArrayList<OTSFuture<BatchWriteRowResult>>();
        int count = 10;
        // 一次性发出10个请求,每个请求写100行数据
        for (int i = 0; i < count; i++) {
            BatchWriteRowRequest request = new BatchWriteRowRequest();
            for (int j = 0; j < 100; j++) {
                RowPutChange rowChange = new RowPutChange(tableName);
                RowPrimaryKey primaryKey = new RowPrimaryKey();
                primaryKey.addPrimaryKeyColumn(COLUMN_GID_NAME, PrimaryKeyValue.fromLong(i * 100 + j));
                primaryKey.addPrimaryKeyColumn(COLUMN_UID_NAME, PrimaryKeyValue.fromLong(j));
                rowChange.setPrimaryKey(primaryKey);
                rowChange.addAttributeColumn(COLUMN_NAME_NAME, ColumnValue.fromString("name" + j));
                rowChange.addAttributeColumn(COLUMN_AGE_NAME, ColumnValue.fromLong(j));

                request.addRowChange(rowChange);
            }
            OTSFuture<BatchWriteRowResult> result = asyncClient.batchWriteRow(request);
            futures.add(result);
        }

        // 等待结果返回
        List<BatchWriteRowResult> results = new ArrayList<BatchWriteRowResult>();
        for (OTSFuture<BatchWriteRowResult> future : futures) {
            try {
                BatchWriteRowResult result = future.get(); // 同步等待结果返回
                results.add(result);
            } catch (OTSException e) {
                e.printStackTrace();
            } catch (ClientException e) {
                e.printStackTrace();
            }
        }

        // 统计返回结果
        int totalSucceedRows = 0;
        int totalFailedRows = 0;
        for (BatchWriteRowResult result : results) {
            totalSucceedRows += result.getSucceedRowsOfPut().size();
            totalFailedRows += result.getFailedRowsOfPut().size();
        }

        System.out.println("Total succeed rows: " + totalSucceedRows);
        System.out.println("Total failed rows: " + totalFailedRows);
    }


案例二:如何实现batch getRange

    private static void batchGetRange(OTSClientAsync asyncClient, String tableName) {
        // 一次性查询多个范围的数据,设置10个任务,每个任务查询100条数据。
        // 每个范围查询的时候设置limit为10,100条数据需要10次请求才能全部查完。
        int count = 10;
        OTSFuture<GetRangeResult>[] futures = new OTSFuture[count];
        for (int i = 0; i < count; i++) {
            futures[i] = sendGetRangeRequest(asyncClient, tableName, i * 100, i * 100 + 100);
        }

        // 检查是否所有范围查询均已做完,若未做完,则继续发送查询请求
        List<Row> allRows = new ArrayList<Row>();
        while (true) {
            boolean completed = true;
            for (int i = 0; i < futures.length; i++) {
                OTSFuture<GetRangeResult> future = futures[i];
                if (future == null) {
                    continue;
                }

                if (future.isDone()) {
                    GetRangeResult result = future.get();
                    allRows.addAll(result.getRows());

                    if (result.getNextStartPrimaryKey() != null) {
                        // 该范围还未查询完毕,需要从nextStart开始继续往下读。
                        long nextStart = result.getNextStartPrimaryKey().getPrimaryKey().get(COLUMN_GID_NAME).asLong();
                        long rangeEnd = i * 100 + 100;
                        futures[i] = sendGetRangeRequest(asyncClient, tableName, nextStart, rangeEnd);
                        completed = false;
                    } else {
                        futures[i] = null; // 若某个范围查询完毕,则将对应future设置为null
                    }
                } else {
                    completed = false;
                }
            }

            if (completed) {
                break;
            } else {
                try {
                    Thread.sleep(10); // 避免busy wait,每次循环完毕后等待一小段时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        // 所有数据全部读出
        System.out.println("Total rows scanned: " + allRows.size());
    }

示例代码可从这里下载。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

阿里云存储基于飞天盘古2.0分布式存储系统,产品多种多样,充分满足用户数据存储和迁移上云需求。

官方博客
链接