WorkManager 两次启动 Worker

Question

我有一组要在后台执行的大任务：

1. 加载数据
2. 解析一堆文件并将它们存储在Room

出于这个原因，我用相同的tag 创建了独特的Worker 链。

class GtfsStaticManager() {
    private val workerManager = WorkManager.getInstance()

    override fun load() {
        val constraints = Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build()

        val inputData = GtfsStaticLoadDataWorker.inputData(staticUrl, cacheDir)
        // 1. Loading data
        val downloadWorkRequest = OneTimeWorkRequest.Builder(GtfsStaticLoadDataWorker::class.java)
            .addTag("GTFS")
            .setConstraints(constraints)
            .setInputData(inputData)
            .build()

        // 2. List of Workers to parse and store data to the Room
        val parseWorkers = GtfsFile.values().map {
            OneTimeWorkRequest.Builder(GtfsStaticParseFileWorker::class.java)
                .setInputData(GtfsStaticParseFileWorker.inputData(it.file, cacheDir + File.separator + "feed"))
                .addTag("GTFS")
                .build()
        }

        workerManager
            .beginUniqueWork("GTFS", ExistingWorkPolicy.KEEP, downloadWorkRequest)
            .then(parseWorkers)
            .enqueue()
    }
}

除了一件事之外，一切都很好：其中一个文件有 400 万条记录，完成它大约需要 10-15 分钟。一段时间后，我注意到它再次入队但第一个作业仍在运行，因此我有 2 个巨大的作业在后台运行，当然我的数据被复制了。

我跟着 codelabs 教程，我错过了什么吗？

下面是我的Worker 解析逻辑：

class GtfsStaticParseFileWorker(
    context: Context,
    workerParameters: WorkerParameters
) : Worker(context, workerParameters) {
    private val fileName: String get() = inputData.getString(FILE_NAME) ?: ""
    private val cacheDir: String get() = inputData.getString(UNZIP_FOLDER) ?: ""

    companion object {
        private const val FILE_NAME = "FILE_NAME"
        private const val UNZIP_FOLDER = "UNZIP_FOLDER"
        fun inputData(fileName: String, cacheDir: String) = Data
            .Builder()
            .putString(FILE_NAME, fileName)
            .putString(UNZIP_FOLDER, cacheDir)
            .build()
    }

    override fun doWork(): Result {
        val db = LvivTransportTrackerDataBase.getUpdateInstance(applicationContext)
        val agencyRepository = AgencyRepository(db.agencyDao())
        val calendarRepository = CalendarRepository(db.calendarDao())
        val calendarDateRepository = CalendarDateRepository(db.calendarDateDao())
        val routeRepository = RouteRepository(db.routeDao())
        val stopTimeRepository = StopTimeRepository(db.stopTimeDao())
        val stopRepository = StopRepository(db.stopDao())
        val tripRepository = TripRepository(db.tripDao())

        val file = File(cacheDir + File.separator + fileName)
        val fileType = GtfsFile.from(fileName) ?: return Result.failure()

        when (fileType) {
            GtfsFile.Agency -> agencyRepository.deleteAll()
            GtfsFile.CalendarDates -> calendarDateRepository.deleteAll()
            GtfsFile.Calendar -> calendarRepository.deleteAll()
            GtfsFile.Routes -> routeRepository.deleteAll()
            GtfsFile.StopTimes -> stopTimeRepository.deleteAll()
            GtfsFile.Stops -> stopRepository.deleteAll()
            GtfsFile.Trips -> tripRepository.deleteAll()
        }

        FileInputStream(file).use { fileInputStream ->
            InputStreamReader(fileInputStream).use inputStreamReader@{ inputStreamReader ->
                val bufferedReader = BufferedReader(inputStreamReader)
                val headers = bufferedReader.readLine()?.split(',') ?: return@inputStreamReader
                var line: String? = bufferedReader.readLine()

                while (line != null) {
                    val mapLine = headers.zip(line.split(',')).toMap()

                    Log.d("GtfsStaticParse", "$fileType: $line")
                    when (fileType) {
                        GtfsFile.Agency -> agencyRepository.create(AgencyEntity(mapLine))
                        GtfsFile.CalendarDates -> calendarDateRepository.create(CalendarDateEntity(mapLine))
                        GtfsFile.Calendar -> calendarRepository.create(CalendarEntity(mapLine))
                        GtfsFile.Routes -> routeRepository.create(RouteEntity(mapLine))
                        GtfsFile.StopTimes -> stopTimeRepository.create(StopTimeEntity(mapLine))
                        GtfsFile.Stops -> stopRepository.create(StopEntity(mapLine))
                        GtfsFile.Trips -> tripRepository.create(TripEntity(mapLine))
                    }

                    line = bufferedReader.readLine()
                }
            }
        }

        return Result.success()
    }
}

附：我的依赖是implementation "android.arch.work:work-runtime:1.0.0"

Answer 1

WorkManager 中的 Worker 类的执行时间限制为 10 分钟。
来自WorkManager guide on how to handle cancellation：

系统指示您的应用因某种原因停止工作。如果您超过 10 分钟的执行期限，就会发生这种情况。该工作计划稍后重试。

在您的情况下，您没有处理工作的停止，但 WorkManager 将忽略任何结果，因为它将作业标记为“已取消”，并会在可能的情况下再次执行。

这可能会导致您遇到双重执行。

在不了解您想要实现的目标的情况下很难提出替代方法，但是，作为一般规则，WorkManager 旨在用于需要保证执行的可延迟任务。

WorkManager documentation 在 1.0 版本之后已经扩展，您可以在那里找到更多信息。

Answer 2

您需要首先检查下载的文件是否存在于下载文件的路径中，然后将字节存储在 SharedPreferences 内部循环中，这些字节是先前从下载路径读取的，一旦您的作业再次启动，然后首先检查下载的字节下一次从那个偏移量开始。

示例代码：

RandomAccessFile seeker = new RandomAccessFile(fname, "r");
seeker.seek(readOffset()); // move to the offset
seeker.readLine(); // and read the String

如果您需要检查相同的数据库条目，则将已完成的行的状态标记为“成功”以便下次跳过。