非线程安全版

  2. import com.amazonaws.annotation.NotThreadSafe;
  3. import com.amazonaws.services.s3.AmazonS3;
  4. import com.amazonaws.services.s3.model.*;
  6. import java.util.*;
  8. @NotThreadSafe
  9. public class S3ObjectsScanner implements Iterable<String> {
  10.     private final static int DEFAULT_MAX_KEYS = 1000;
  12.     /**
  13.      * S3客户端对象
  14.      */
  15.     private AmazonS3 client;
  16.     /**
  17.      * 桶名
  18.      */
  19.     private String bucketName;
  20.     /**
  21.      * 起始索引
  22.      */
  23.     private String startPrefix;
  24.     /**
  25.      * 终止索引
  26.      */
  27.     private String endPrefix;
  28.     /**
  29.      * key的最大缓存个数
  30.      */
  31.     private int maxKeys;
  33.     public S3ObjectsScanner() {
  34.         this.maxKeys = DEFAULT_MAX_KEYS;
  35.     }
  37.     public S3ObjectsScanner(AmazonS3 client, String bucketName) {
  38.         this();
  39.         this.client = client;
  40.         this.bucketName = bucketName;
  41.     }
  43.     @NotThreadSafe
  44.     private class Itr implements Iterator<String> {
  45.         private final ListObjectsV2Request request;
  46.         private final Queue<String> queue;
  47.         private final String endPrefix;
  48.         private boolean lastPage = false;
  50.         Itr() {
  51.             this.endPrefix = S3ObjectsScanner.this.endPrefix;
  52.             request = new ListObjectsV2Request()
  53.                     .withBucketName(bucketName)
  54.                     .withStartAfter(startPrefix)
  55.                     .withMaxKeys(maxKeys);
  56.             queue = new ArrayDeque<>(maxKeys);
  57.         }
  59.         @Override
  60.         public synchronized boolean hasNext() {
  61.             if (queue.isEmpty() && !lastPage) {
  62.                 loadMore();
  63.             }
  65.             return !queue.isEmpty();
  66.         }
  68.         @Override
  69.         public synchronized String next() {
  70.             if (!hasNext()) {
  71.                 throw new NoSuchElementException();
  72.             }
  74.             return queue.poll();
  75.         }
  77.         private void loadMore() {
  78.             // 发送请求
  79.             ListObjectsV2Result result = client.listObjectsV2(request);
  80.             // 准备下一次请求
  81.             prepareNextReq(result);
  82.             // 提取Keys
  83.             extraKeys(result.getObjectSummaries());
  84.         }
  86.         private void prepareNextReq(ListObjectsV2Result result) {
  87.             // 检查是否还有剩余数据
  88.             if (result.isTruncated()) {
  89.                 request.setContinuationToken(result.getNextContinuationToken());
  90.             } else {
  91.                 // 没有剩余数据，置位
  92.                 lastPage = true;
  93.             }
  94.         }
  96.         private void extraKeys(List<S3ObjectSummary> summaries) {
  97.             for (S3ObjectSummary summary : summaries) {
  98.                 String key = summary.getKey();
  99.                 // 检查是否到达结束的位置
  100.                 if (isOver(key)) {
  101.                     // 结束，标志位置位，结束循环
  102.                     lastPage = true;
  103.                     return;
  104.                 }
  105.                 // 未结束，入队
  106.                 queue.offer(key);
  107.             }
  108.         }
  110.         private boolean isOver(String key) {
  111.             // 如果没有指定“终止前缀”，无论什么key都不算结束
  112.             if (endPrefix == null) {
  113.                 return false;
  114.             }
  116.             // 如果key小于“终止前缀”，也不算结束
  117.             if (endPrefix.compareTo(key) > 0) {
  118.                 return false;
  119.             }
  121.             return true;
  122.         }
  124.     }
  126.     @Override
  127.     public Iterator<String> iterator() {
  128.         return new Itr();
  129.     }
  131.     public AmazonS3 getClient() {
  132.         return client;
  133.     }
  135.     public S3ObjectsScanner withClient(AmazonS3 client) {
  136.         this.client = client;
  137.         return this;
  138.     }
  140.     public String getBucketName() {
  141.         return bucketName;
  142.     }
  144.     public S3ObjectsScanner withBucketName(String bucketName) {
  145.         this.bucketName = bucketName;
  146.         return this;
  147.     }
  149.     public String getStartPrefix() {
  150.         return startPrefix;
  151.     }
  153.     public S3ObjectsScanner withStartPrefix(String startPrefix) {
  154.         this.startPrefix = startPrefix;
  155.         return this;
  156.     }
  158.     public String getEndPrefix() {
  159.         return endPrefix;
  160.     }
  162.     public S3ObjectsScanner withEndPrefix(String endPrefix) {
  163.         this.endPrefix = endPrefix;
  164.         return this;
  165.     }
  167.     public int getMaxKeys() {
  168.         return maxKeys;
  169.     }
  171.     public S3ObjectsScanner withMaxKeys(int maxKeys) {
  172.         this.maxKeys = maxKeys;
  173.         return this;
  174.     }
  175. }