MaxCompute用户指南：数据上传下载：批量数据通道SDK示例：多线程下载示例

1. import java.io.IOException;
2. import java.util.ArrayList;
3. import java.util.Date;
4. import java.util.List;
5. import java.util.concurrent.Callable;
6. import java.util.concurrent.ExecutionException;
7. import java.util.concurrent.ExecutorService;
8. import java.util.concurrent.Executors;
9. import java.util.concurrent.Future;
11. import com.aliyun.odps.Column;
12. import com.aliyun.odps.Odps;
13. import com.aliyun.odps.PartitionSpec;
14. import com.aliyun.odps.TableSchema;
15. import com.aliyun.odps.account.Account;
16. import com.aliyun.odps.account.AliyunAccount;
17. import com.aliyun.odps.data.Record;
18. import com.aliyun.odps.data.RecordReader;
19. import com.aliyun.odps.tunnel.TableTunnel;
20. import com.aliyun.odps.tunnel.TableTunnel.DownloadSession;
21. import com.aliyun.odps.tunnel.TunnelException;
23. class DownloadThread implements Callable<Long> {
24.          private long id;
25.          private RecordReader recordReader;
26.          private TableSchema tableSchema;
28.          public DownloadThread(int id,
29.                          RecordReader recordReader, TableSchema tableSchema) {
30.                  this.id = id;
31.                  this.recordReader = recordReader;
32.                  this.tableSchema = tableSchema;
33.          }
35.          @Override
36.          public Long call() {
37.                  Long recordNum = 0L;
38.                  try {
39.                          Record record;
40.                          while ((record = recordReader.read()) != null) {
41.                                  recordNum++;
42.                                  System.out.print("Thread " + id + "\t");
43.                                  consumeRecord(record, tableSchema);
44.                          }
45.                          recordReader.close();
46.                  } catch (IOException e) {
47.                          e.printStackTrace();
48.                  }
49.                  return recordNum;
50.          }
52.          private static void consumeRecord(Record record, TableSchema schema) {
53.                  for (int i = 0; i < schema.getColumns().size(); i++) {
54.                          Column column = schema.getColumn(i);
55.                          String colValue = null;
56.                          switch (column.getType()) {
57.                          case BIGINT: {
58.                                  Long v = record.getBigint(i);
59.                                  colValue = v == null ? null : v.toString();
60.                                  break;
61.                          }
62.                          case BOOLEAN: {
63.                                  Boolean v = record.getBoolean(i);
64.                                  colValue = v == null ? null : v.toString();
65.                                  break;
66.                          }
67.                          case DATETIME: {
68.                                  Date v = record.getDatetime(i);
69.                                  colValue = v == null ? null : v.toString();
70.                                  break;
71.                          }
72.                          case DOUBLE: {
73.                                  Double v = record.getDouble(i);
74.                                  colValue = v == null ? null : v.toString();
75.                                  break;
76.                          }
77.                          case STRING: {
78.                                  String v = record.getString(i);
79.                                  colValue = v == null ? null : v.toString();
80.                                  break;
81.                          }
82.                          default:
83.                                  throw new RuntimeException("Unknown column type: "
84.                                                  + column.getType());
85.                          }
86.                          System.out.print(colValue == null ? "null" : colValue);
87.                          if (i != schema.getColumns().size())
88.                                  System.out.print("\t");
89.                  }
90.                  System.out.println();
91.          }
93. }
95. public class DownloadThreadSample {
96.          private static String accessId = "<your access id>";
97.          private static String accessKey = "<your access Key>";
99.          private static String odpsUrl = "http://service.odps.aliyun.com/api";
102.          private static String project = "<your project>";
103.          private static String table = "<your table name>";
104.          private static String partition = "<your partition spec>";
106.          private static int threadNum = 10;
108.          public static void main(String args[]) {
109.                  Account account = new AliyunAccount(accessId, accessKey);
110.                  Odps odps = new Odps(account);
111.                  odps.setEndpoint(odpsUrl);
112.                  odps.setDefaultProject(project);
113.                  TableTunnel tunnel = new TableTunnel(odps);
114.                  PartitionSpec partitionSpec = new PartitionSpec(partition);
115.                  DownloadSession downloadSession;
116.                  try {
117.                          downloadSession = tunnel.createDownloadSession(project, table,
118.                                          partitionSpec);
120.                          System.out.println("Session Status is : "
121.                                          + downloadSession.getStatus().toString());
123.                          long count = downloadSession.getRecordCount();
124.                          System.out.println("RecordCount is: " + count);
126.                          ExecutorService pool = Executors.newFixedThreadPool(threadNum);
127.                          ArrayList<Callable<Long>> callers = new ArrayList<Callable<Long>>();
129.                          long start = 0;
130.                          long step = count / threadNum;
131.                          for (int i = 0; i < threadNum - 1; i++) {
132.                                  RecordReader recordReader = downloadSession.openRecordReader(
133.                                                  step * i, step);
134.                                  callers.add(new DownloadThread( i, recordReader, downloadSession.getSchema()));
135.                          }
136.                          RecordReader recordReader = downloadSession.openRecordReader(step * (threadNum - 1), count
137.                                          - ((threadNum - 1) * step));
138.                          callers.add(new DownloadThread( threadNum - 1, recordReader, downloadSession.getSchema()));
140.                          Long downloadNum = 0L;
141.                          List<Future<Long>> recordNum = pool.invokeAll(callers);
142.                          for (Future<Long> num : recordNum)
143.                                  downloadNum += num.get();
144.                          System.out.println("Record Count is: " + downloadNum);
145.                          pool.shutdown();
147.                  } catch (TunnelException e) {
148.                          e.printStackTrace();
149.                  } catch (IOException e) {
150.                          e.printStackTrace();
151.                  } catch (InterruptedException e) {
152.                          e.printStackTrace();
153.                  } catch (ExecutionException e) {
154.                          e.printStackTrace();
155.                  }
156.          }
157. }

注意：
对于 Tunnel Endpoint，支持指定或者不指定。

• 
  如果指定，按照指定的 Endpoint 下载。
• 
  如果不指定，支持按照自动路由下载。