All files / services memory-queue.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
/**
 * 内存队列 - 当 Redis 不可用时的 fallback
 * 使用内存 Map 存储任务，支持基本的队列功能
 */
 
type JobData = Record<string, any>;
type JobCallback = (job: any) => Promise<any>;
 
interface MemoryJob {
  id: string;
  data: JobData;
  status: 'waiting' | 'active' | 'completed' | 'failed';
  result?: any;
  error?: string;
}
 
class MemoryQueue {
  private jobs: Map<string, MemoryJob> = new Map();
  private waitingJobs: string[] = [];
  private processing: Set<string> = new Set();
  private idCounter: number = 0;
  private concurrency: number = 3;
  private handler: JobCallback | null = null;
  private started: boolean = false;
 
  async add(name: string, data: JobData): Promise<string> {
    const id = `${name}_${++this.idCounter}_${Date.now()}`;
    const job: MemoryJob = {
      id,
      data,
      status: 'waiting',
    };
    this.jobs.set(id, job);
    this.waitingJobs.push(id);
    console.log(`[MemoryQueue] 任务添加: ${name}#${id}`);
 
    // 如果已经有 handler，开始处理
    if (this.handler && !this.started) {
      this.startProcessing();
    } else if (this.started && this.processing.size < this.concurrency) {
      // 已经在运行且有空闲槽位，立即处理
      setImmediate(() => this.processLoop());
    }
 
    return id;
  }
 
  process(concurrency: number, handler: JobCallback): void {
    this.concurrency = concurrency;
    this.handler = handler;
    this.startProcessing();
  }
 
  private startProcessing() {
    if (this.started || !this.handler) return;
    this.started = true;
 
    // 异步处理队列
    setImmediate(() => this.processLoop());
  }
 
  private async processLoop() {
    while (this.waitingJobs.length > 0 && this.processing.size < this.concurrency) {
      const jobId = this.waitingJobs.shift();
      if (!jobId) break;
 
      const job = this.jobs.get(jobId);
      if (!job || job.status !== 'waiting') continue;
 
      this.processing.add(jobId);
      job.status = 'active';
 
      try {
        console.log(`[MemoryQueue] 开始处理任务: ${jobId}`);
        const mockJob = {
          id: jobId,
          data: job.data,
          progress: (p: number) => { /* 进度回调 */ }
        };
        const result = await this.handler(mockJob);
        job.status = 'completed';
        job.result = result;
        console.log(`[MemoryQueue] 任务完成: ${jobId}`);
      } catch (error: any) {
        job.status = 'failed';
        job.error = error.message;
        console.error(`[MemoryQueue] 任务失败: ${jobId}`, error.message);
      } finally {
        this.processing.delete(jobId);
        // 继续处理下一个
        setImmediate(() => this.processLoop());
      }
    }
 
    // 如果还有任务没处理完，继续循环
    if (this.waitingJobs.length > 0 || this.processing.size > 0) {
      setTimeout(() => this.processLoop(), 100);
    }
  }
 
  on(event: string, callback: any): void {
    // 事件处理
  }
 
  async close(): Promise<void> {
    this.jobs.clear();
    this.waitingJobs = [];
    this.processing.clear();
    this.started = false;
  }
 
  // 检查是否有待处理任务
  hasPendingJobs(): boolean {
    return this.waitingJobs.length > 0 || this.processing.size > 0;
  }
}
 
// 导出单例
export const memoryQueue = new MemoryQueue();