Android WorkManager中的异步工作者

Question

谷歌最近宣布了新的WorkManager架构组件.通过WorkManager在doWork()课堂上实现,可以很容易地安排同步工作,但是如果我想在后台进行一些异步工作呢？例如,我想使用Retrofit进行网络服务调用.我知道我可以发出同步网络请求,但它会阻塞线程并且感觉不对.有没有解决方案,或者目前还不支持？

Answer 1

我使用了countdownlatch并等待它达到0,这只会在异步回调更新后才会发生.看到这段代码:

public WorkerResult doWork() {

        final WorkerResult[] result = {WorkerResult.RETRY};
        CountDownLatch countDownLatch = new CountDownLatch(1);
        FirebaseFirestore db = FirebaseFirestore.getInstance();

        db.collection("collection").whereEqualTo("this","that").get().addOnCompleteListener(task -> {
            if(task.isSuccessful()) {
                task.getResult().getDocuments().get(0).getReference().update("field", "value")
                        .addOnCompleteListener(task2 -> {
                            if (task2.isSuccessful()) {
                                result[0] = WorkerResult.SUCCESS;
                            } else {
                                result[0] = WorkerResult.RETRY;
                            }
                            countDownLatch.countDown();
                        });
            } else {
                result[0] = WorkerResult.RETRY;
                countDownLatch.countDown();
            }
        });

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return result[0];

    }

Answer 2

仅供参考,现在有ListenableWorker,它是异步的.

编辑:以下是一些示例用法的片段.我删除了大块代码,我认为这些代码不是说明性的,因此很可能会出现一两个小错误.

这适用于采用String photoKey,从服务器检索元数据,执行某些压缩工作,然后上载压缩照片的任务.这发生在主线程之外.以下是我们发送工作请求的方式:

private void compressAndUploadFile(final String photoKey) {
    Data inputData = new Data.Builder()
            .putString(UploadWorker.ARG_PHOTO_KEY, photoKey)
            .build();
    Constraints constraints = new Constraints.Builder()
            .setRequiredNetworkType(NetworkType.CONNECTED)
            .build();
    OneTimeWorkRequest request = new OneTimeWorkRequest.Builder(UploadWorker.class)
            .setInputData(inputData)
            .setConstraints(constraints)
            .build();
    WorkManager.getInstance().enqueue(request);
}

在UploadWorker中:

public class UploadWorker extends ListenableWorker {
    private static final String TAG = "UploadWorker";
    public static final String ARG_PHOTO_KEY = "photo-key";

    private String mPhotoKey;

    /**
     * @param appContext   The application {@link Context}
     * @param workerParams Parameters to setup the internal state of this worker
     */
    public UploadWorker(@NonNull Context appContext, @NonNull WorkerParameters workerParams) {
        super(appContext, workerParams);
        mPhotoKey = workerParams.getInputData().getString(ARG_PHOTO_KEY);
    }

    @NonNull
    @Override
    public ListenableFuture<Payload> onStartWork() {
        SettableFuture<Payload> future = SettableFuture.create();
        Photo photo = getPhotoMetadataFromServer(mPhotoKey).addOnCompleteListener(task -> {
            if (!task.isSuccessful()) {
                Log.e(TAG, "Failed to retrieve photo metadata", task.getException());
                future.setException(task.getException());
                return;
            }
            MyPhotoType photo = task.getResult();
            File file = photo.getFile();
            Log.d(TAG, "Compressing " + photo);
            MyImageUtil.compressImage(file, MyConstants.photoUploadConfig).addOnCompleteListener(compressionTask -> {
                if (!compressionTask.isSuccessful()) {
                    Log.e(TAG, "Could not parse " + photo + " as an image.", compressionTask.getException());
                    future.set(new Payload(Result.FAILURE));
                    return;
                }
                byte[] imageData = compressionTask.getResult();
                Log.d(TAG, "Done compressing " + photo);
                UploadUtil.uploadToServer(photo, imageData);
                future.set(new Payload(Result.SUCCESS));
            });
        });
        return future;
    }
}

Answer 3

每个WorkManager文档:

默认情况下,WorkManager在后台线程上运行其操作.如果您已经在后台线程上运行并且需要对WorkManager进行同步(阻塞)调用,请使用synchronous()来访问此类方法.

因此,如果您不使用synchronous(),您可以安全地执行同步网络呼叫doWork().从设计角度来看,这也是一种更好的方法,因为回调很混乱.

也就是说,如果你真的想从中激活异步作业doWork(),你需要暂停执行线程并在使用wait/notify机制(或其他一些线程管理机制Semaphore)完成异步作业时恢复它.在大多数情况下,我不建议这样做.

作为旁注,WorkManager处于非常早期的alpha状态.

Answer 4

如果您正在谈论asynchronus工作,您可以将您的工作转移到RxJava Observables/Singles.

有一组运算符喜欢.blockingGet()或.blockingFirst() 转换Observable<T>为阻塞T

Worker在后台线程上执行所以不要担心NetworkOnMainThreadException.

Answer 5

借助协程的强大功能，您可以doWork()像这样“同步”：

用于获取位置的挂起方法（异步）：

private suspend fun getLocation(): Location = suspendCoroutine { continuation ->
    val mFusedLocationClient = LocationServices.getFusedLocationProviderClient(appContext)
    mFusedLocationClient.lastLocation.addOnSuccessListener {
        continuation.resume(it)
    }.addOnFailureListener {
        continuation.resumeWithException(it)
    }
}

调用示例doWork()：

override fun doWork(): Result {
    val loc = runBlocking {
        getLocation()
    }
    val latitude = loc.latitude
}

2021 更新：您现在可以使用CoroutineWorker，它具有 suspenddoWork()方法。

class MySuspendWorker(private val appContext: Context, workerParams: WorkerParameters) : CoroutineWorker(appContext, workerParams) {
    override suspend fun doWork(): Result {
        //do your async work
    }
}

Answer 6

我使用过BlockingQueue，它简化了线程同步和线程之间传递结果，您只需要一个对象

private var disposable = Disposables.disposed()

private val completable = Completable.fromAction { 
        //do some heavy computation
    }.subscribeOn(Schedulers.computation()) // you will do the work on background thread

override fun doWork(): Result {
    val result = LinkedBlockingQueue<Result>()

    disposable = completable.subscribe(
            { result.put(Result.SUCCESS) },
            { result.put(Result.RETRY) }
    )

    return try {
        result.take() //need to block this thread untill completable has finished
    } catch (e: InterruptedException) {
        Result.RETRY
    }
}

如果您的 Worker 已停止，也不要忘记释放资源，这是主要优势，.blockingGet()因为现在您可以正确地自由取消您的 Rx 任务。

override fun onStopped(cancelled: Boolean) {
    disposable.dispose()
}