@ agendajs/ redis-后端

重写后端 议程项目 pub/Sub 支持实时工作处理的工作调度器.

特征

• 议程任务的全部 Redis 存储后端
• 使用 Redis Pub/Sub 的实时工作通知
• 高效的工作索引与排序集
• 原子工作与WATCH/MULTI/EXEC
• 连接集合通过 ioredis 图书馆
• 带有完整类型定义的脚本类型支持

安装

npm install @agendajs/redis-backend
# or
pnpm add @agendajs/redis-backend
# or
yarn add @agendajs/redis-backend

使用量

基本用途

import { Agenda } from 'agenda';
import { RedisBackend } from '@agendajs/redis-backend';

// Create agenda with Redis backend
const agenda = new Agenda({
  backend: new RedisBackend({
    connectionString: 'redis://localhost:6379'
  })
});

// Define jobs
agenda.define('send-email', async (job) => {
  const { to, subject, body } = job.attrs.data;
  await sendEmail(to, subject, body);
});

// Start processing
await agenda.start();

// Schedule jobs
await agenda.every('5 minutes', 'send-email', { to: '[email protected]', subject: 'Hello', body: 'World' });
await agenda.schedule('in 10 minutes', 'send-email', { to: '[email protected]', subject: 'Hi', body: 'There' });

配置选项

import { RedisBackend } from '@agendajs/redis-backend';

const backend = new RedisBackend({
  // Redis connection string (required unless redis/redisOptions is provided)
  connectionString: 'redis://localhost:6379',

  // Or use Redis client options (creates a new client)
  redisOptions: {
    host: 'localhost',
    port: 6379,
    password: 'secret',
    db: 0
  },

  // Key prefix for all Redis keys (default: 'agenda:')
  keyPrefix: 'agenda:',

  // Channel name for Pub/Sub notifications (default: 'agenda:notifications')
  channelName: 'agenda:notifications',

  // Name to identify this Agenda instance (stored as lastModifiedBy)
  name: 'worker-1',

  // Sort order for job queries
  sort: {
    nextRunAt: 'asc',
    priority: 'desc'
  }
});

使用已有的 Redis 客户端

如果您的应用程序已经有Redis客户端,您可以直接通过. 客户会的 没有 当议程断开时关闭 :

import { Redis } from 'ioredis';
import { RedisBackend } from '@agendajs/redis-backend';

// Your app's existing Redis client
const redis = new Redis('redis://localhost:6379');

const backend = new RedisBackend({ redis });

// After agenda.stop(), your Redis client is still usable

如何运作

存储

后端存储使用 Redis 数据结构的工作 :

• 哈谢斯 (单位:千美元){prefix}job:{id}) - 将工作数据作为密钥值对存储
• 设定 (单位:千美元){prefix}jobs:all- 跟踪所有工作ID
• 设定 (单位:千美元){prefix}jobs:by_name:{name}) - 快速检索按名称列出的工作索引
• 排序的集 (单位:千美元){prefix}jobs:by_next_run_at) - 为高效调度,在下个运行前索引任务
• 字符串 (单位:千美元){prefix}jobs:single:{name}) - 跟踪单类型工作,用于高端操作

实时通知

瑞迪斯拥有内置的酒吧/分公司能力. 当一项工作被保存时,会向所有签署《议程》的事例发布通知,立即启动工作处理,而不必等待下一次投票间隔.

这意味着:

• 执行职务的延迟度较低
• 减少投票费用
• 二. 有效的跨进程协调

分布式锁定

后端使用 Redis WATCH/MULTI/EXEC 交易在解剖上锁定任务进行处理,防止多个"议程"实例的重复执行.

任务日志

RedisBackend 包含内置 RedisJobLogger 储存结构化的工作生命周期事件 使用Redis排序的集和散列。 伐木机是轻量级的,只在第一次使用时创建了它的键.

自动使用

当您在议事日程中启用日志时, 后端的内置登录器会被自动使用 :

const agenda = new Agenda({
  backend: new RedisBackend({ connectionString: 'redis://localhost:6379' }),
  logging: true  // Uses RedisJobLogger automatically
});

独立使用

您还可以使用 RedisJobLogger 独立——例如,在使用不同的后端存储时登录到 Redis :

import { RedisJobLogger } from '@agendajs/redis-backend';
import Redis from 'ioredis';

const logger = new RedisJobLogger({
  redis: new Redis('redis://localhost:6379')
});

// Use with any backend
import { MongoBackend } from '@agendajs/mongo-backend';
const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  logging: logger
});

见 议程项目研究 用于完整记录文件.

持久性

默认情况下,Redis将数据保存在内存中. 为了与《议程》一起生产,请配置Redis的持久性:

# In redis.conf
appendonly yes           # Enable AOF persistence
appendfsync everysec     # Sync to disk every second

或使用 RDB 快照 :

save 900 1      # Save after 900 seconds if at least 1 key changed
save 300 10     # Save after 300 seconds if at least 10 keys changed
save 60 10000   # Save after 60 seconds if at least 10000 keys changed

对于最大耐久性,同时启用AOF和RDB.

测试

测试需要一个 Redis 服务器 。 最简单的方法是使用多克:

# Start Redis container and run tests
pnpm test:docker

# Or manually:
pnpm docker:up      # Start Redis container
pnpm test           # Run tests
pnpm docker:down    # Stop container

您也可以使用已有的 Redis 服务器 :

REDIS_TEST_URL=redis://localhost:6379 pnpm test

可用的脚本

所需资源

• 节点.js 18.0.0
• 雷迪丝 >= 6.0
• 议程项目6.0.0

许可证

麻省理工学院