并发控制 #

async function fetchAll(urls: string[]): Promise<Response[]> {
  const promises = urls.map(url => fetch(url));
  return Promise.all(promises);
}

const responses = await fetchAll([
  "https://api.example.com/users",
  "https://api.example.com/posts"
]);

const results = await Promise.allSettled([
  fetch("/api/users"),
  fetch("/api/posts"),
  fetch("/api/comments")
]);

results.forEach((result, i) => {
  if (result.status === "fulfilled") {
    console.log(`请求 ${i} 成功`);
  } else {
    console.log(`请求 ${i} 失败: ${result.reason}`);
  }
});

async function fetchWithTimeout(
  url: string,
  timeout: number
): Promise<Response> {
  return Promise.race([
    fetch(url),
    new Promise<never>((_, reject) => 
      setTimeout(() => reject(new Error("Timeout")), timeout)
    )
  ]);
}

async function fetchFastest(urls: string[]): Promise<Response> {
  const promises = urls.map(url => fetch(url));
  return Promise.any(promises);
}

const response = await fetchFastest([
  "https://mirror1.example.com/data",
  "https://mirror2.example.com/data",
  "https://mirror3.example.com/data"
]);

async function limitConcurrency<T, R>(
  items: T[],
  limit: number,
  fn: (item: T) => Promise<R>
): Promise<R[]> {
  const results: R[] = [];
  const executing: Promise<void>[] = [];
  
  for (const [index, item] of items.entries()) {
    const promise = fn(item).then(result => {
      results[index] = result;
    });
    
    executing.push(promise);
    
    if (executing.length >= limit) {
      await Promise.race(executing);
      executing.splice(
        executing.findIndex(p => p === promise),
        1
      );
    }
  }
  
  await Promise.all(executing);
  return results;
}

// 使用示例
const urls = Array.from({ length: 100 }, (_, i) => `https://api.example.com/${i}`);

const results = await limitConcurrency(urls, 5, async (url) => {
  const response = await fetch(url);
  return response.json();
});

class ConcurrencyPool {
  private running = 0;
  private queue: Array<() => void> = [];
  
  constructor(private limit: number) {}
  
  async run<T>(fn: () => Promise<T>): Promise<T> {
    while (this.running >= this.limit) {
      await new Promise<void>(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    try {
      return await fn();
    } finally {
      this.running--;
      const next = this.queue.shift();
      if (next) next();
    }
  }
}

// 使用示例
const pool = new ConcurrencyPool(3);

const results = await Promise.all(
  urls.map(url => pool.run(() => fetch(url).then(r => r.json())))
);

class TaskQueue<T> {
  private queue: T[] = [];
  private processing = false;
  
  constructor(
    private processor: (item: T) => Promise<void>,
    private concurrency: number = 1
  ) {}
  
  add(item: T): void {
    this.queue.push(item);
    this.process();
  }
  
  private async process(): Promise<void> {
    if (this.processing) return;
    this.processing = true;
    
    while (this.queue.length > 0) {
      const batch = this.queue.splice(0, this.concurrency);
      await Promise.all(batch.map(item => this.processor(item)));
    }
    
    this.processing = false;
  }
}

interface PriorityTask<T> {
  item: T;
  priority: number;
}

class PriorityQueue<T> {
  private tasks: PriorityTask<T>[] = [];
  
  add(item: T, priority: number = 0): void {
    const task = { item, priority };
    const index = this.tasks.findIndex(t => t.priority < priority);
    
    if (index === -1) {
      this.tasks.push(task);
    } else {
      this.tasks.splice(index, 0, task);
    }
  }
  
  get(): T | undefined {
    const task = this.tasks.shift();
    return task?.item;
  }
  
  get length(): number {
    return this.tasks.length;
  }
}

async function batchFetch(
  urls: string[],
  batchSize: number = 5
): Promise<Response[]> {
  const results: Response[] = [];
  
  for (let i = 0; i < urls.length; i += batchSize) {
    const batch = urls.slice(i, i + batchSize);
    const responses = await Promise.all(batch.map(url => fetch(url)));
    results.push(...responses);
  }
  
  return results;
}

async function fetchWithRetry(
  url: string,
  maxRetries: number = 3
): Promise<Response> {
  let lastError: Error | undefined;
  
  for (let i = 0; i < maxRetries; i++) {
    try {
      const response = await fetch(url);
      if (response.ok) return response;
      throw new Error(`HTTP ${response.status}`);
    } catch (error) {
      lastError = error as Error;
      await new Promise(resolve => setTimeout(resolve, 1000 * (i + 1)));
    }
  }
  
  throw lastError;
}

async function fetchAllWithRetry(
  urls: string[],
  concurrency: number = 5
): Promise<Response[]> {
  return limitConcurrency(urls, concurrency, fetchWithRetry);
}

async function pipeline<T, R>(
  items: T[],
  concurrency: number,
  ...stages: Array<(item: T) => Promise<T>>
): Promise<R[]> {
  const results: R[] = [];
  
  await limitConcurrency(items, concurrency, async (item) => {
    let value: any = item;
    for (const stage of stages) {
      value = await stage(value);
    }
    results.push(value);
  });
  
  return results;
}

// 使用示例
const processed = await pipeline(
  urls,
  5,
  async (url) => fetch(url),
  async (response) => response.json(),
  async (data) => transform(data)
);