74 lines
3.0 KiB
TypeScript
74 lines
3.0 KiB
TypeScript
|
// services/worker/src/main.ts —— photo worker 进程入口（Task 21）。构建产物
|
|||
|
|
// dist/main.js 由根 dev:worker / 生产启动命令运行；env 一律来自进程环境（根
|
|||
|
|
// dotenv-cli 注入），此处不读文件。启动顺序：Prisma + MinIO 适配器 → 两条专用
|
|||
|
|
// ioredis 连接（Worker 阻塞命令须独占连接；Queue/sweep 入队共用另一条）→
|
|||
|
|
// BullMQ Worker（并发 2）+ 30s 补偿扫描。优雅退出：SIGINT/SIGTERM 先停扫描，
|
|||
|
|
// worker.close()（等待在飞 job）后逐个关闭连接（Task 19 gotcha：BullMQ 不接管
|
|||
|
|
// 外部连接，须显式 quit；未监听的 error 事件会崩进程，全部挂日志监听）。
|
|||
|
|
import { Queue } from 'bullmq';
|
|||
|
|
import IORedis from 'ioredis';
|
|||
|
|
import { pino, type Logger } from 'pino';
|
|||
|
|
import { PrismaClient } from '@photo/db-cn';
|
|||
|
|
import { QUEUE_PHOTO_PROCESS } from '@photo/contracts';
|
|||
|
|
import { createStorageFromEnv } from '@photo/storage';
|
|||
|
|
import { createPhotoProcessWorker } from './process.processor';
|
|||
|
|
import { startSweepInterval } from './sweep';
|
|||
|
|
|
|||
|
|
const SWEEP_INTERVAL_MS = 30_000;
|
|||
|
|
|
|||
|
|
function createConnection(logger: Logger, label: string): IORedis {
|
|||
|
|
const connection = new IORedis(process.env.REDIS_URL_CN ?? 'redis://localhost:6379', {
|
|||
|
|
maxRetriesPerRequest: null, // BullMQ 阻塞命令要求：禁用 per-request 重试上限
|
|||
|
|
});
|
|||
|
|
connection.on('error', (err: Error) => {
|
|||
|
|
logger.warn({ label, err: err.message }, 'redis 连接异常（ioredis 自动重连）');
|
|||
|
|
});
|
|||
|
|
return connection;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async function main(): Promise<void> {
|
|||
|
|
const logger = pino({ level: process.env.LOG_LEVEL ?? 'info' });
|
|||
|
|
const prisma = new PrismaClient();
|
|||
|
|
const storage = createStorageFromEnv();
|
|||
|
|
const workerConnection = createConnection(logger, 'worker');
|
|||
|
|
const queueConnection = createConnection(logger, 'queue');
|
|||
|
|
const sweepQueue = new Queue(QUEUE_PHOTO_PROCESS, { connection: queueConnection });
|
|||
|
|
const worker = createPhotoProcessWorker({
|
|||
|
|
prisma,
|
|||
|
|
storage,
|
|||
|
|
connection: workerConnection,
|
|||
|
|
logger,
|
|||
|
|
});
|
|||
|
|
const sweep = startSweepInterval(SWEEP_INTERVAL_MS, { prisma, queue: sweepQueue, logger });
|
|||
|
|
logger.info(
|
|||
|
|
{ queue: QUEUE_PHOTO_PROCESS, concurrency: 2, sweepIntervalMs: SWEEP_INTERVAL_MS },
|
|||
|
|
'photo worker started',
|
|||
|
|
);
|
|||
|
|
|
|||
|
|
let stopping = false;
|
|||
|
|
const shutdown = async (signal: string): Promise<void> => {
|
|||
|
|
if (stopping) return;
|
|||
|
|
stopping = true;
|
|||
|
|
logger.info({ signal }, 'photo worker shutting down');
|
|||
|
|
sweep.stop();
|
|||
|
|
try {
|
|||
|
|
await worker.close(); // 等待在飞 job 完成
|
|||
|
|
await sweepQueue.close();
|
|||
|
|
await workerConnection.quit().catch(() => undefined);
|
|||
|
|
await queueConnection.quit().catch(() => undefined);
|
|||
|
|
await prisma.$disconnect();
|
|||
|
|
} catch (err) {
|
|||
|
|
logger.error(
|
|||
|
|
{ err: err instanceof Error ? err.message : String(err) },
|
|||
|
|
'photo worker 关闭异常',
|
|||
|
|
);
|
|||
|
|
} finally {
|
|||
|
|
process.exit(0);
|
|||
|
|
}
|
|||
|
|
};
|
|||
|
|
process.once('SIGINT', () => void shutdown('SIGINT'));
|
|||
|
|
process.once('SIGTERM', () => void shutdown('SIGTERM'));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
void main();
|