Angular实现请求并发数控制与排队机制的方案咨询

Got it, let's fix this request queuing issue that's causing your ping requests to get blocked and users to be logged out. Here's a clean, Angular-native approach to control concurrent HTTP requests globally—works no matter which component or service is firing requests.

We'll use an HTTP Interceptor to hook into every outgoing request, paired with a dedicated service to manage a request queue and enforce a maximum concurrent request limit (like 4 or 5). When the number of active requests hits the limit, new requests get added to a queue. As soon as an active request finishes (success or failure), we pull the next request from the queue and execute it.

1. Create a Concurrency Control Service #

This service will track active requests, hold the queue, and handle executing requests in order.

import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class RequestConcurrencyService {
  private activeRequests = 0;
  private maxConcurrentRequests = 4; // Default limit, can be updated
  private requestQueue: Array<() => Observable<any>> = [];

  // Allow dynamic updates to the max concurrent limit
  setMaxConcurrentRequests(limit: number): void {
    this.maxConcurrentRequests = limit;
    this.processQueue(); // Kick off queued requests if we have space now
  }

  addRequest(requestFn: () => Observable<any>): Observable<any> {
    return new Observable((observer) => {
      const execute = () => {
        this.activeRequests++;
        requestFn().subscribe({
          next: val => observer.next(val),
          error: err => observer.error(err),
          complete: () => {
            this.activeRequests--;
            observer.complete();
            this.processQueue(); // Trigger next request in queue
          }
        });
      };

      // If we haven't hit the limit, execute immediately
      if (this.activeRequests < this.maxConcurrentRequests) {
        execute();
      } else {
        // Otherwise, add to queue
        this.requestQueue.push(execute);
      }
    });
  }

  private processQueue(): void {
    // Keep processing until we hit the limit or queue is empty
    while (this.activeRequests < this.maxConcurrentRequests && this.requestQueue.length) {
      const nextRequest = this.requestQueue.shift();
      nextRequest?.();
    }
  }
}

2. Build the HTTP Interceptor #

This interceptor will route every request through our concurrency service.

import { Injectable } from '@angular/core';
import { HttpRequest, HttpHandler, HttpEvent, HttpInterceptor } from '@angular/common/http';
import { Observable } from 'rxjs';
import { RequestConcurrencyService } from './request-concurrency.service';

@Injectable()
export class ConcurrencyInterceptor implements HttpInterceptor {
  constructor(private concurrencyService: RequestConcurrencyService) {}

  intercept(request: HttpRequest<unknown>, next: HttpHandler): Observable<HttpEvent<unknown>> {
    // Pass the request handler to our concurrency service
    return this.concurrencyService.addRequest(() => next.handle(request));
  }
}

3. Register the Interceptor #

Add the interceptor to your app module so it applies globally.

import { NgModule } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';
import { HttpClientModule, HTTP_INTERCEPTORS } from '@angular/common/http';
import { ConcurrencyInterceptor } from './concurrency.interceptor';
import { RequestConcurrencyService } from './request-concurrency.service';

@NgModule({
  declarations: [/* Your components here */],
  imports: [BrowserModule, HttpClientModule],
  providers: [
    RequestConcurrencyService,
    {
      provide: HTTP_INTERCEPTORS,
      useClass: ConcurrencyInterceptor,
      multi: true // Allows multiple interceptors
    }
  ],
  bootstrap: [/* Your root component */]
})
export class AppModule { }

4. Set Your Desired Concurrent Limit #

You can set the max concurrent requests anywhere in your app (e.g., root component, config service):

import { Component, OnInit } from '@angular/core';
import { RequestConcurrencyService } from './request-concurrency.service';

@Component({
  selector: 'app-root',
  template: `<router-outlet></router-outlet>`
})
export class AppComponent implements OnInit {
  constructor(private concurrencyService: RequestConcurrencyService) {}

  ngOnInit(): void {
    // Set to 5 concurrent requests
    this.concurrencyService.setMaxConcurrentRequests(5);
  }
}

Since your ping requests are time-sensitive (they keep users logged in), you can exclude them from the queue to let them bypass the concurrency limit. Update the interceptor:

intercept(request: HttpRequest<unknown>, next: HttpHandler): Observable<HttpEvent<unknown>> {
  // Check if this is your ping request (adjust the URL match to your actual endpoint)
  if (request.url.includes('/api/user/ping')) {
    return next.handle(request); // Skip queue, send immediately
  }
  return this.concurrencyService.addRequest(() => next.handle(request));
}

This ensures your ping requests never get stuck behind other traffic, eliminating the logout issue.

内容的提问来源于stack exchange，提问作者Krishna