如何用同步块序列化异步调用

Question

我使用Firebase作为我的后端，在客户端(Android)方面，我试图按顺序下载一堆图像。我将迭代器包装在synchronized块中，等待下载每个图像。

private Object mLock = new Object();
private void downloadImages() {
   List<StorageReference> storageReferences = getStorageReferences();
   synchronized (mLock) {
      // Iterate trough all image references 
      for (StorageReference sr : storageReferences) {
         sr.getBytes(ONE_MB_BUFFER).addOnCompleteListener(new OnCompleteListener<byte[]>() {
            @Override
            public void onComplete(Task<byte[]> task) {
               if (task.isSuccessful()) {
                  // Success, image downloaded
               }
               // Notify, that we have downloaded the image
               synchronized (mLock) {
                  mLock.notify();
               }
            }
         });
         // Await until we acquire the lock
         try {
            mLock.wait();
         }
         catch (InterruptedException e) {
            e.printStackTrace();
         }
      }
   }
}

addOnCompleteListener中的回调不被调用。实际上，整个线程都是锁定的。

是否有其他方法来排队下载任务?像单线程执行器服务?

Answer 1

我最终使用ExecutorService与newSingleThreadExecutor。如果您需要其他自定义，如超时，您可能需要使用newScheduledThreadPool。您可以创建一个线程池并同时执行多个线程。

public class ImageDownloadService extends IntentService {
   @Override
   protected void onHandleIntent(Intent intent) {
      downloadImages();
   }  
   private void downloadImages() {
      ExecutorService executor = Executors.newSingleThreadExecutor();
      List<StorageReference> storageReferences = getStorageReferences();
      for (StorageReference sr : storageReferences) {
         Future<byte[]> future = executor.submit(new FutureImageResult(sr));
         byte[] data = null;
         try {
            data = future.get();
         } catch (InterruptedException e) {
            e.printStackTrace();
         } catch (ExecutionException e) {
            e.printStackTrace();
         }
         if (data != null && data.length > 0) {
            // Image downloaded successfully
         }
      }
   }   
}

提交给executor服务的future。

public class FutureImageResult implements Callable<byte[]> {
    private StorageReference mStorageReference;
    private boolean mIsFailure;
    public FutureImageResult(StorageReference storageReference) {
        mStorageReference = storageReference;
    }
    @Override
    public byte[] call() throws Exception {
        Task<byte[]> task = mStorageReference.getBytes(1024 * 1024);
        task.addOnFailureListener(new OnFailureListener() {
            @Override
            public void onFailure(@NonNull Exception e) {
                mIsFailure = true;
            }
        });
        while (!task.isComplete() || mIsFailure);
        byte[] data = task.getResult();
        return data;
    }
}

Answer 2

另一种方法是使用BlockingQueue。

总体思路是:

• 你启动一个线程，它不断地从队列中轮询，下载给定的图像并重新开始
• 将所有url放入队列
• 一个监听器与线程或每个图像相关联

更少的代码是:

final BlockingQueue<URL> queue = new LinkedBlockingQueue<>();
new Thread(new Runnable() {
    @Override
    public void run() {
        while (true) {
            URL url = queue.poll();
            // Download the image and notify the listener               
        }
    }
}).start();

这个线程可以由Service启动，这样它就可以继续运行，而不绑定到UI。活动可以绑定服务来与之交互。

Answer 3

您也可以使用CountDownLatch来锁定您的工作线程，直到操作完成。像这样:

private void downloadImages() {
   List<StorageReference> storageReferences = getStorageReferences();
   CountDownLatch waitForDownload = new CountDownLatch(storageReferences.size());
   // Iterate trough all image references 
   for (StorageReference sr : storageReferences) {
         sr.getBytes(ONE_MB_BUFFER).addOnCompleteListener(new OnCompleteListener<byte[]>() {
            @Override
            public void onComplete(Task<byte[]> task) {
               // Notify, that we have downloaded the image and continue
               waitForDownload.countDown();
            }
         });
   }
   // Lock until we download all images
   waitForDownload.await();
   // Continue with the rest of your serialized work having all images downloaded
   ...
}

参考文献:CountDownLatch javadoc.