HaHafeng
fa72beea6c
Major Changes: - Implement Platform-Only architecture pattern (unified task management) - Add PostgresCacheAdapter for unified caching (platform_schema.app_cache) - Add PgBossQueue for job queue management (platform_schema.job) - Implement CheckpointService using job.data (generic for all modules) - Add intelligent threshold-based dual-mode processing (THRESHOLD=50) - Add task splitting mechanism (auto chunk size recommendation) - Refactor ASL screening service with smart mode selection - Refactor DC extraction service with smart mode selection - Register workers for ASL and DC modules Technical Highlights: - All task management data stored in platform_schema.job.data (JSONB) - Business tables remain clean (no task management fields) - CheckpointService is generic (shared by all modules) - Zero code duplication (DRY principle) - Follows 3-layer architecture principle - Zero additional cost (no Redis needed, save 8400 CNY/year) Code Statistics: - New code: ~1750 lines - Modified code: ~500 lines - Test code: ~1800 lines - Documentation: ~3000 lines Testing: - Unit tests: 8/8 passed - Integration tests: 2/2 passed - Architecture validation: passed - Linter errors: 0 Files: - Platform layer: PostgresCacheAdapter, PgBossQueue, CheckpointService, utils - ASL module: screeningService, screeningWorker - DC module: ExtractionController, extractionWorker - Tests: 11 test files - Docs: Updated 4 key documents Status: Phase 1-7 completed, Phase 8-9 pending
127 lines
2.8 KiB
TypeScript
127 lines
2.8 KiB
TypeScript
import { JobQueue } from './types.js'
|
||
import { MemoryQueue } from './MemoryQueue.js'
|
||
import { PgBossQueue } from './PgBossQueue.js'
|
||
|
||
/**
|
||
* 任务队列工厂类
|
||
*
|
||
* 根据环境变量自动选择队列实现：
|
||
* - QUEUE_TYPE=memory: 使用MemoryQueue（内存队列）
|
||
* - QUEUE_TYPE=pgboss: 使用PgBossQueue（Postgres队列）
|
||
* - QUEUE_TYPE=database: 别名，指向pgboss
|
||
*
|
||
* 零代码切换：
|
||
* - 本地开发：不配置QUEUE_TYPE，默认使用memory
|
||
* - Postgres-Only架构：配置QUEUE_TYPE=pgboss
|
||
* - 多实例部署：配置QUEUE_TYPE=pgboss（自动负载均衡）
|
||
*
|
||
* @example
|
||
* ```typescript
|
||
* import { jobQueue } from '@/common/jobs'
|
||
*
|
||
* // 业务代码不关心具体实现
|
||
* const job = await jobQueue.push('asl:screening', { projectId: 123 })
|
||
* ```
|
||
*/
|
||
export class JobFactory {
|
||
private static instance: JobQueue | null = null
|
||
|
||
/**
|
||
* 获取任务队列实例（单例模式）
|
||
*/
|
||
static getInstance(): JobQueue {
|
||
if (!this.instance) {
|
||
this.instance = this.createQueue()
|
||
}
|
||
return this.instance
|
||
}
|
||
|
||
/**
|
||
* 创建任务队列
|
||
*/
|
||
private static createQueue(): JobQueue {
|
||
const queueType = process.env.QUEUE_TYPE || 'memory'
|
||
|
||
switch (queueType) {
|
||
case 'memory':
|
||
return this.createMemoryQueue()
|
||
|
||
case 'pgboss':
|
||
case 'database': // 别名
|
||
return this.createPgBossQueue()
|
||
|
||
default:
|
||
console.warn(`[JobFactory] Unknown QUEUE_TYPE: ${queueType}, fallback to memory`)
|
||
return this.createMemoryQueue()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 创建内存队列
|
||
*/
|
||
private static createMemoryQueue(): MemoryQueue {
|
||
console.log('[JobFactory] Using MemoryQueue')
|
||
|
||
const queue = new MemoryQueue()
|
||
|
||
// 定期清理已完成的任务（避免内存泄漏）
|
||
if (process.env.NODE_ENV !== 'test') {
|
||
setInterval(() => {
|
||
queue.cleanup()
|
||
}, 60 * 60 * 1000) // 每小时清理一次
|
||
}
|
||
|
||
return queue
|
||
}
|
||
|
||
/**
|
||
* 创建PgBoss队列
|
||
*/
|
||
private static createPgBossQueue(): PgBossQueue {
|
||
const databaseUrl = process.env.DATABASE_URL
|
||
|
||
if (!databaseUrl) {
|
||
throw new Error(
|
||
'[JobFactory] DATABASE_URL is required when QUEUE_TYPE=pgboss'
|
||
)
|
||
}
|
||
|
||
console.log('[JobFactory] Using PgBossQueue (Postgres-Only架构)')
|
||
|
||
const queue = new PgBossQueue(databaseUrl, 'platform_schema')
|
||
|
||
// 启动队列（异步）
|
||
queue.start().catch(err => {
|
||
console.error('[JobFactory] Failed to start PgBossQueue:', err)
|
||
})
|
||
|
||
// 定期清理缓存中的已完成任务
|
||
if (process.env.NODE_ENV !== 'test') {
|
||
setInterval(() => {
|
||
queue.cleanup()
|
||
}, 60 * 60 * 1000) // 每小时清理一次
|
||
}
|
||
|
||
return queue
|
||
}
|
||
|
||
/**
|
||
* 重置实例（用于测试）
|
||
*/
|
||
static reset(): void {
|
||
this.instance = null
|
||
}
|
||
}
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|