Flutter后台任务管理插件pg_job_queue的使用

Flutter后台任务管理插件pg_job_queue的使用

pg_job_queue 是一个基于 PostgreSQL 的简单任务队列。通过这个插件，你可以在 Flutter 应用程序中管理后台任务。

配置

唯一需要配置的是 PostgreSQL 连接对象。最基本的配置如下所示：

import 'package:pg_job_queue/pg_job_queue.dart';
import 'package:postgres/postgres.dart';

void main() async {
  /// 在本地运行时，启动 PostgreSQL：
  /// `docker run -p 5432:5432 -e POSTGRES_PASSWORD=postgres postgres`
  final connection = await Connection.open(
      Endpoint(
        host: 'localhost',
        database: 'postgres',
        username: 'postgres',
        password: 'postgres',
      ),
      settings: ConnectionSettings(sslMode: SslMode.disable));

  /// 创建一个新的任务队列，使用默认设置。
  final jobs = PgJobQueue(connection);

  /// 初始化数据库表。这是幂等的。
  await jobs.init();
  
  /// 此时，任务队列已经准备好使用。
  
  /// 完成后关闭连接。
  await connection.close();
}

上述代码创建了一个新的任务队列，并使用默认设置。init 方法会在数据库中创建必要的表，而且它是幂等的，所以可以多次调用而不会出现问题。每次应用程序启动时调用 init 方法都是安全的。当发布新的次要版本时，init 方法应该能够自动将数据库模式更新到新版本。

默认情况下，任务队列使用 public 模式中的 jobs 表。默认队列名称为 default。可以通过向 PgJobQueue 构造函数传递相关的参数来更改此配置。生成的任务 ID 是 UUID v4。可以通过向 PgJobQueue 构造函数传递自定义的 idGenerator 函数来更改这一点。

调度任务

要调度任务，使用 schedule 方法。它返回已调度任务的 ID。提供将传递给任务处理程序的 JSON 负载。也可以提供自定义的 queue，如果你想使用不同的队列。每个任务还可以分配一个 priority 值。默认优先级是 0。允许负值。优先级值较高的任务会首先执行。

final jobId = await jobs.schedule(
  payload: {'key': 'value'},
  queue: 'default',
  priority: 0,
);

当任务被调度时，它会被插入到 jobs 表中，并且状态为 scheduled。

检查任务状态

要检查任务的状态，使用 fetch 方法。它返回一个 Job 对象。

final job = await jobs.fetch(jobId);
print(job.status);

处理任务

要获取队列中的下一个任务，使用 acquire 方法。它返回一个 Job 对象。可以提供自定义的 queue 来从不同的队列中获取任务。另一个可选参数是 worker，这是一个任意字符串，用于标识正在处理任务的工作器。这在调试和监控方面很有用。

完成处理后，使用 complete 或 fail 方法将结果发送回任务队列。

final job = await jobs.acquire(queue: 'emails', worker: 'worker1');
print(job.payload);
await jobs.complete(job.id, result: {'key': 'value'});

清理

要从数据库中删除已完成的任务，使用 deleteCompleted 方法。它删除所有比指定时间更早的已完成任务。可以使用 deleteFailed 参数删除失败的任务。可以指定 limit 来分批删除任务。建议定期运行此方法以提高性能。

await jobs.deleteCompleted(Duration(days: 30), limit: 1000);

统计信息

一些基本统计信息可以通过 countByQueueByStatus 方法获得。它返回一个包含每个队列和状态的作业数量的映射。

final stats = await jobs.countByQueueByStatus();
print(stats);

性能考虑因素

任务队列设计为简单高效。每个操作都使用单个数据库连接和单个事务。但是，对于较大的数据量，你可以考虑以下几点：

• 限制命名队列的数量。共享同一表的队列会影响彼此的性能。
• 为每个队列使用单独的表。这可以通过向 PgJobQueue 构造函数传递自定义的 schema 和 table 来实现。
• 为每个队列使用单独的数据库连接。

示例代码

以下是一个完整的示例代码，展示了如何使用 pg_job_queue 插件：

import 'package:pg_job_queue/pg_job_queue.dart';
import 'package:postgres/postgres.dart';

void main() async {
  /// 在本地运行时，启动 PostgreSQL：
  /// `docker run -p 5432:5432 -e POSTGRES_PASSWORD=postgres postgres`
  final connection = await Connection.open(
      Endpoint(
        host: 'localhost',
        database: 'postgres',
        username: 'postgres',
        password: 'postgres',
      ),
      settings: ConnectionSettings(sslMode: SslMode.disable));

  /// 创建一个新的任务队列，使用默认设置。
  final jobs = PgJobQueue(connection);

  /// 初始化数据库表。这是幂等的。
  await jobs.init();

  /// 调度一些任务。
  for (var i = 0; i < 10; i++) {
    await jobs.schedule({'name': 'Job $i'});
  }

  await Future.wait(['Alice', 'Bob', 'Chris'].map((worker) async {
    print('Worker $worker started');
    while (true) {
      final job = await jobs.acquire(worker: worker);
      if (job == null) {
        print('All done. $worker shutting down');
        break;
      }
      print('Worker $worker processing job ${job.id} (${job.payload})');
      await jobs.complete(job.id);
    }
  }));

  await connection.close();
}