devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

面试官：假如有几十个请求，如何去控制并发？_使用线程池整个服务限制请求数量

作者：笔触狂放9 | 2024-05-24 13:50:21

赞

踩

使用线程池整个服务限制请求数量

控制并发请求是一个重要的问题，特别是在面对高并发情况时，合理地管理请求可以有效地维护系统的稳定性和性能。以下是一些常见的方法来控制并发请求：

1. 线程池：使用线程池来管理并发请求，通过限制线程数量和队列大小，可以控制同时处理的请求数量，避免系统因过多的请求而崩溃或性能下降。

2. 信号量：使用信号量来限制并发请求的数量，通过控制信号量的数量来控制允许同时处理的请求数量，超过限制的请求将被阻塞或拒绝。

3. 限流算法：采用限流算法来控制请求的流量，例如令牌桶算法、漏桶算法等，可以根据系统的处理能力和资源情况来限制请求的发送速率，避免系统被过多的请求压垮。

4. 分布式锁：在分布式系统中，使用分布式锁来控制对共享资源的并发访问，通过加锁和解锁操作来保证同一时刻只有一个请求可以访问共享资源，避免并发冲突。

5. 缓存：利用缓存来减轻对后端系统的压力，将频繁请求的数据缓存起来，减少对数据库或其他后端服务的访问次数，提高系统的响应速度和性能。

6. 服务降级：在高并发情况下，可以根据系统的负载情况对服务进行降级，暂时关闭一些不太重要或资源密集型的功能，以保证系统的核心功能正常运行。

1. 线程池

没错，线程池是控制并发请求的一种有效方式。下面是一些关于线程池的详细说明：

线程池的作用：

  • 线程池可以预先创建一定数量的线程，并将它们保存在池中以备重用。
  • 当有请求到达时，线程池可以从池中获取一个空闲线程来处理请求，而不是每次请求都创建新的线程，从而避免了线程频繁创建和销毁的开销。

线程池的核心参数：

  • 线程数量：线程池中允许同时存在的线程数量，通过控制线程数量可以控制同时处理的请求数量。
  • 队列大小：当线程池中的线程都被占用时，新的请求会被放入队列中等待处理。队列的大小可以限制同时排队等待的请求数量，避免队列过长导致的资源浪费或性能下降。

线程池的优点：

  • 降低线程创建和销毁的开销：线程池中的线程可以被重复利用，避免了频繁创建和销毁线程的开销。
  • 控制并发度：通过调整线程池的大小和队列的大小，可以有效地控制同时处理的请求数量，避免系统因过载而崩溃或性能下降。
  • 提高响应速度：线程池中的线程可以立即响应新的请求，而不需要等待新线程的创建，提高了系统的响应速度。

线程池的注意事项：

  • 合理设置线程池的大小和队列大小，避免过小导致无法处理大量请求或过大导致资源浪费。
  • 注意线程池中线程的生命周期管理，及时关闭不再需要的线程，避免资源占用过多。

总的来说，线程池是一种常见且有效的并发控制机制，可以帮助系统更好地处理并发请求，提高系统的稳定性和性能。

以下是一个简单的Java代码示例，演示如何使用线程池来控制并发请求：

  1. import java.util.concurrent.ExecutorService;
  2. import java.util.concurrent.Executors;
  3.  
  4. public class ThreadPoolExample {
  5.     public static void main(String[] args) {
  6.         // 创建一个固定大小的线程池，大小为3
  7.         ExecutorService executor = Executors.newFixedThreadPool(3);
  8.  
  9.         // 模拟10个请求
  10.         for (int i = 0; i < 10; i++) {
  11.             // 提交任务给线程池执行
  12.             executor.submit(new RequestTask(i + 1));
  13.         }
  14.  
  15.         // 关闭线程池
  16.         executor.shutdown();
  17.     }
  18. }
  19.  
  20. // 请求任务类
  21. class RequestTask implements Runnable {
  22.     private int requestId;
  23.  
  24.     public RequestTask(int requestId) {
  25.         this.requestId = requestId;
  26.     }
  27.  
  28.     @Override
  29.     public void run() {
  30.         // 模拟请求处理过程
  31.         System.out.println("处理请求 " + requestId + "，线程：" + Thread.currentThread().getName());
  32.  
  33.         try {
  34.             // 模拟请求处理耗时
  35.             Thread.sleep(1000);
  36.         } catch (InterruptedException e) {
  37.             e.printStackTrace();
  38.         }
  39.  
  40.         System.out.println("请求 " + requestId + " 处理完成");
  41.     }
  42. }

在这个示例中，我们使用了Executors.newFixedThreadPool()方法创建了一个固定大小为3的线程池。然后，我们模拟了10个请求，每个请求是一个RequestTask对象，每个请求处理耗时1秒。我们将这些请求提交给线程池执行。由于线程池的大小是3，所以只有3个请求能够同时被处理，其余的请求会被放入线程池的队列中等待处理。最后，我们调用executor.shutdown()关闭线程池。

这个例子展示了如何使用线程池来控制并发请求的数量，避免系统因过多请求而崩溃或性能下降。

2. 信号量

信号量是一种经典的并发控制工具，可以用来限制并发请求的数量。下面是一个简单的Java代码示例，演示如何使用信号量来限制并发请求的数量：

  1. import java.util.concurrent.Semaphore;
  2.  
  3. public class SemaphoreExample {
  4.     public static void main(String[] args) {
  5.         // 创建一个信号量，限制同时处理的请求数量为3
  6.         Semaphore semaphore = new Semaphore(3);
  7.  
  8.         // 模拟10个请求
  9.         for (int i = 0; i < 10; i++) {
  10.             // 启动一个线程处理请求
  11.             new Thread(new RequestTask(i + 1, semaphore)).start();
  12.         }
  13.     }
  14. }
  15.  
  16. // 请求任务类
  17. class RequestTask implements Runnable {
  18.     private int requestId;
  19.     private Semaphore semaphore;
  20.  
  21.     public RequestTask(int requestId, Semaphore semaphore) {
  22.         this.requestId = requestId;
  23.         this.semaphore = semaphore;
  24.     }
  25.  
  26.     @Override
  27.     public void run() {
  28.         try {
  29.             // 获取信号量许可
  30.             semaphore.acquire();
  31.             // 模拟请求处理过程
  32.             System.out.println("处理请求 " + requestId + "，线程：" + Thread.currentThread().getName());
  33.  
  34.             // 模拟请求处理耗时
  35.             Thread.sleep(1000);
  36.  
  37.             System.out.println("请求 " + requestId + " 处理完成");
  38.         } catch (InterruptedException e) {
  39.             e.printStackTrace();
  40.         } finally {
  41.             // 释放信号量许可
  42.             semaphore.release();
  43.         }
  44.     }
  45. }

在这个示例中，我们使用Semaphore类创建了一个信号量，限制了同时处理的请求数量为3。然后，我们模拟了10个请求，每个请求是一个RequestTask对象，每个请求处理耗时1秒。我们启动了10个线程来处理这些请求，在每个线程中，首先调用semaphore.acquire()获取一个信号量许可，如果当前已经达到了限制的请求数量，则该线程会被阻塞直到有许可可用。当请求处理完成后，调用semaphore.release()释放信号量许可，表示该请求处理完成，其他被阻塞的请求可以继续执行。

这个例子展示了如何使用信号量来限制并发请求的数量，保护系统免受过载的影响。

3. 限流算法

限流算法是一种常见的控制请求流量的方法，它可以根据系统的处理能力和资源情况来限制请求的发送速率，防止系统被过多的请求压垮。常见的限流算法包括令牌桶算法和漏桶算法。

下面分别介绍这两种算法：

  1. 令牌桶算法：

    • 令牌桶算法通过维护一个令牌桶来控制请求的发送速率。令牌桶中存放着固定数量的令牌，每个令牌代表一个请求的许可。当有请求到达时，需要从令牌桶中获取一个令牌，如果令牌桶中没有足够的令牌，则请求会被暂时阻塞或拒绝。
    • 令牌桶算法可以平滑地限制请求的发送速率，并且可以应对突发流量，因为如果令牌桶中的令牌未被消耗完，那么多余的令牌会被保留供后续使用。

  2. 漏桶算法：

    • 漏桶算法通过一个固定容量的漏桶来控制请求的发送速率。当请求到达时，会被放入漏桶中，然后以固定的速率从漏桶中流出。如果漏桶已满，则新的请求会被丢弃或者等待下一个时间窗口。
    • 漏桶算法可以平滑地控制请求的发送速率，但是无法处理突发流量，因为漏桶容量固定，无法应对突发的大量请求。

这些限流算法可以根据系统的实际情况选择合适的算法来控制请求流量，保护系统免受过载的影响。在实际应用中，常常结合使用这些算法来实现更加灵活和可靠的限流策略。

令牌桶算法实现示例：

  1. import java.util.concurrent.ArrayBlockingQueue;
  2. import java.util.concurrent.BlockingQueue;
  3.  
  4. public class TokenBucket {
  5.     private final BlockingQueue<Object> tokenBucket;
  6.     private final int capacity;
  7.  
  8.     public TokenBucket(int capacity) {
  9.         this.capacity = capacity;
  10.         tokenBucket = new ArrayBlockingQueue<>(capacity);
  11.         // 初始化令牌桶，向其中添加令牌
  12.         for (int i = 0; i < capacity; i++) {
  13.             tokenBucket.offer(new Object());
  14.         }
  15.         // 启动一个线程，定期往令牌桶中添加新的令牌
  16.         new Thread(this::refillToken).start();
  17.     }
  18.  
  19.     public void refillToken() {
  20.         while (true) {
  21.             try {
  22.                 Thread.sleep(1000); // 每秒往令牌桶中添加一个令牌
  23.                 tokenBucket.offer(new Object());
  24.             } catch (InterruptedException e) {
  25.                 e.printStackTrace();
  26.             }
  27.         }
  28.     }
  29.  
  30.     public void getToken() {
  31.         try {
  32.             tokenBucket.take(); // 从令牌桶中获取一个令牌，如果没有可用的令牌则阻塞
  33.             System.out.println("Token obtained, processing request...");
  34.             // 处理请求
  35.         } catch (InterruptedException e) {
  36.             e.printStackTrace();
  37.         }
  38.     }
  39.  
  40.     public static void main(String[] args) {
  41.         TokenBucket tokenBucket = new TokenBucket(10); // 令牌桶容量为10
  42.         // 模拟10个请求
  43.         for (int i = 0; i < 10; i++) {
  44.             new Thread(() -> tokenBucket.getToken()).start();
  45.         }
  46.     }
  47. }

漏桶算法实现示例：

  1. import java.util.concurrent.TimeUnit;
  2.  
  3. public class LeakyBucket {
  4.     private final int capacity; // 漏桶容量
  5.     private final int rate; // 漏水速率，单位：毫秒
  6.     private int water; // 当前水量
  7.     private long lastLeakTime; // 上一次漏水时间
  8.  
  9.     public LeakyBucket(int capacity, int rate) {
  10.         this.capacity = capacity;
  11.         this.rate = rate;
  12.         this.water = 0;
  13.         this.lastLeakTime = System.currentTimeMillis();
  14.         // 启动一个线程，定期执行漏水操作
  15.         new Thread(this::leak).start();
  16.     }
  17.  
  18.     public synchronized void leak() {
  19.         while (true) {
  20.             long now = System.currentTimeMillis();
  21.             // 计算距离上一次漏水过去的时间
  22.             long elapsedTime = now - lastLeakTime;
  23.             // 计算当前应该漏掉的水量
  24.             int leakAmount = (int) (elapsedTime / rate);
  25.             // 更新上一次漏水时间
  26.             lastLeakTime = now;
  27.             // 更新当前水量
  28.             water = Math.max(0, water - leakAmount);
  29.             try {
  30.                 // 每隔一段时间执行漏水操作
  31.                 TimeUnit.MILLISECONDS.sleep(rate);
  32.             } catch (InterruptedException e) {
  33.                 e.printStackTrace();
  34.             }
  35.         }
  36.     }
  37.  
  38.     public synchronized boolean acceptRequest() {
  39.         // 检查漏桶中的水量是否已满
  40.         if (water < capacity) {
  41.             // 漏桶未满，允许请求通过，并增加水量
  42.             water++;
  43.             System.out.println("Request accepted, processing...");
  44.             // 处理请求
  45.             return true;
  46.         } else {
  47.             // 漏桶已满，拒绝请求
  48.             System.out.println("Bucket is full, request rejected.");
  49.             return false;
  50.         }
  51.     }
  52.  
  53.     public static void main(String[] args) {
  54.         LeakyBucket leakyBucket = new LeakyBucket(10, 1000); // 漏桶容量为10，漏水速率为每秒1个请求
  55.         // 模拟20个请求
  56.         for (int i = 0; i < 20; i++) {
  57.             new Thread(() -> leakyBucket.acceptRequest()).start();
  58.             try {
  59.                 Thread.sleep(500); // 间隔500毫秒发送一个请求
  60.             } catch (InterruptedException e) {
  61.                 e.printStackTrace();
  62.             }
  63.         }
  64.     }
  65. }

以上示例展示了令牌桶算法和漏桶算法在Java中的简单实现。这些实现仅用于演示目的，实际应用中可能需要根据具体需求进行更复杂的实现和优化。

4. 分布式锁

分布式锁是在分布式系统中用来控制对共享资源的访问的一种机制。在分布式环境中，多个节点之间需要协调访问共享资源，以确保数据的一致性和可靠性。分布式锁通过加锁和解锁操作来实现对共享资源的互斥访问，保证同一时刻只有一个请求可以对资源进行操作，从而避免并发冲突。

常见的分布式锁实现方式包括：

  1. 基于数据库的分布式锁： 使用数据库的事务特性和行级锁来实现分布式锁。通过在数据库中创建一个特定的表来存储锁信息，利用数据库的事务机制来确保对锁的操作的原子性和一致性。

  2. 基于缓存的分布式锁： 使用分布式缓存服务如Redis或Memcached来实现分布式锁。通过在缓存中存储锁的键值对，利用缓存服务的原子性操作来实现加锁和解锁操作。

  3. 基于ZooKeeper的分布式锁： 使用ZooKeeper这样的分布式协调服务来实现分布式锁。通过在ZooKeeper的节点上创建临时顺序节点，并利用其特性来实现锁的竞争和释放。

  4. 基于分布式一致性算法的分布式锁： 如基于Paxos或Raft等一致性算法来实现分布式锁。这种方式通常需要借助一致性算法的实现来确保分布式锁的正确性和可靠性。

选择合适的分布式锁实现方式取决于具体的场景需求、性能要求和系统架构等因素。每种实现方式都有其优缺点，开发人员需要根据实际情况进行选择和权衡。

5. 缓存

以下是基于Redis实现的简单分布式锁的Java代码示例：

  1. import redis.clients.jedis.Jedis;
  2.  
  3. public class DistributedLock {
  4.  
  5.     private static final String LOCK_KEY = "my_lock";
  6.     private static final int EXPIRE_TIME = 30000; // 锁的过期时间，单位毫秒
  7.     private static final int WAIT_TIME = 1000; // 获取锁的等待时间，单位毫秒
  8.  
  9.     private Jedis jedis;
  10.  
  11.     public DistributedLock() {
  12.         // 连接Redis服务器
  13.         jedis = new Jedis("localhost");
  14.     }
  15.  
  16.     public boolean acquireLock() {
  17.         long start = System.currentTimeMillis();
  18.         try {
  19.             // 循环尝试获取锁，直到超时
  20.             while (System.currentTimeMillis() - start < WAIT_TIME) {
  21.                 // 尝试获取锁
  22.                 if (jedis.setnx(LOCK_KEY, "locked") == 1) {
  23.                     // 设置锁的过期时间，防止锁无法释放造成死锁
  24.                     jedis.pexpire(LOCK_KEY, EXPIRE_TIME);
  25.                     return true; // 成功获取锁
  26.                 }
  27.                 // 休眠一段时间后重试
  28.                 Thread.sleep(100);
  29.             }
  30.         } catch (InterruptedException e) {
  31.             Thread.currentThread().interrupt();
  32.         }
  33.         return false; // 获取锁超时
  34.     }
  35.  
  36.     public void releaseLock() {
  37.         // 释放锁
  38.         jedis.del(LOCK_KEY);
  39.     }
  40.  
  41.     public static void main(String[] args) {
  42.         DistributedLock lock = new DistributedLock();
  43.         if (lock.acquireLock()) {
  44.             // 成功获取锁，执行业务逻辑
  45.             System.out.println("Lock acquired, executing business logic...");
  46.             try {
  47.                 Thread.sleep(5000); // 模拟业务逻辑执行时间
  48.             } catch (InterruptedException e) {
  49.                 e.printStackTrace();
  50.             } finally {
  51.                 lock.releaseLock(); // 执行完业务逻辑后释放锁
  52.                 System.out.println("Lock released.");
  53.             }
  54.         } else {
  55.             // 获取锁失败，执行其他处理逻辑
  56.             System.out.println("Failed to acquire lock.");
  57.         }
  58.     }
  59. }

以上示例使用了Redis的SETNX命令来尝试获取锁，SETNX命令用于设置一个键的值，如果该键不存在，则设置成功，返回1；如果该键已经存在，则设置失败，返回0。利用这一特性，可以通过SETNX命令来实现分布式锁的获取。

当获取锁成功后，使用PEXPIRE命令给锁设置一个过期时间，以防止锁因为某些异常情况无法释放而造成死锁。在释放锁时，通过DEL命令删除键来释放锁。

请注意，这只是一个简单的示例，实际中可能需要考虑更多的异常情况和优化。

6. 服务降级

服务降级是一种在系统压力过大或异常情况下，通过暂时关闭或限制某些功能或服务来保证系统的核心功能可用性和稳定性的策略。在高并发情况下，如果系统无法及时响应所有请求，可能会导致系统性能下降甚至崩溃，为了避免这种情况发生，可以通过服务降级来保护系统的核心功能。

服务降级的优点包括：

  1. 保证核心功能可用性： 通过暂时关闭或限制非关键功能，确保系统核心功能能够正常运行，提高系统的稳定性和可用性。

  2. 降低系统压力： 关闭一些不太重要或资源密集型的功能，减少系统的负载，从而降低系统的压力，保证核心功能的响应速度。

  3. 提高系统鲁棒性： 在面对异常情况或攻击时，通过服务降级可以防止系统因负载过大而崩溃，提高系统的抗压能力和鲁棒性。

服务降级的实施可以通过以下方式：

  1. 关闭非关键功能： 暂时关闭一些不太重要或非核心的功能，例如一些统计信息展示、广告推荐等，以释放系统资源。

  2. 限流控制： 对于一些可能导致系统负载过高的接口或功能，可以通过限流控制来限制请求的访问速率，保护系统不受过多请求的影响。

  3. 降级响应： 对于某些接口或功能，可以返回简化的响应或错误提示，而不是完整的数据或功能，以降低服务器的压力。

  4. 优先级调整： 对于不同的请求或用户，可以根据其重要性或优先级，调整服务的响应速度或资源分配情况，确保关键用户或业务能够优先得到响应。

总的来说，服务降级是保障系统稳定性和可用性的重要手段之一，在系统设计和开发过程中需要考虑如何合理地实施服务降级策略，以应对不同的负载情况和异常场景。

下面是一个简单的Java代码示例，演示如何实现基本的服务降级策略，当系统压力过大时，暂时关闭某些功能：

  1. public class ServiceDegradation {
  2.  
  3.     private static boolean isSystemUnderPressure = false;
  4.  
  5.     // 模拟核心功能的服务
  6.     public static void coreFunctionality() {
  7.         if (isSystemUnderPressure) {
  8.             System.out.println("System is under pressure. Core functionality degraded.");
  9.             return;
  10.         }
  11.         // 实际核心功能的实现
  12.         System.out.println("Executing core functionality...");
  13.     }
  14.  
  15.     // 模拟关闭非关键功能以降级服务
  16.     public static void nonCriticalFunctionality() {
  17.         if (isSystemUnderPressure) {
  18.             System.out.println("System is under pressure. Non-critical functionality degraded.");
  19.             return;
  20.         }
  21.         // 实际非关键功能的实现
  22.         System.out.println("Executing non-critical functionality...");
  23.     }
  24.  
  25.     // 模拟系统压力过大的情况
  26.     public static void systemUnderPressure() {
  27.         isSystemUnderPressure = true;
  28.         System.out.println("System is under pressure. Degraded services activated.");
  29.     }
  30.  
  31.     public static void main(String[] args) {
  32.         // 正常情况下执行核心功能
  33.         coreFunctionality();
  34.  
  35.         // 正常情况下执行非关键功能
  36.         nonCriticalFunctionality();
  37.  
  38.         // 模拟系统压力过大，激活服务降级
  39.         systemUnderPressure();
  40.  
  41.         // 在系统压力过大时执行核心功能
  42.         coreFunctionality();
  43.  
  44.         // 在系统压力过大时执行非关键功能
  45.         nonCriticalFunctionality();
  46.     }
  47. }

在这个简单的示例中，我们定义了一个ServiceDegradation类，其中包含了两个模拟服务：coreFunctionality()代表系统的核心功能，nonCriticalFunctionality()代表一个非关键功能。通过systemUnderPressure()方法模拟系统压力过大的情况，激活服务降级。

在正常情况下，调用coreFunctionality()和nonCriticalFunctionality()时，会执行它们的实际功能。但是，在系统压力过大时，调用这两个方法时会输出相应的服务降级提示，而不会执行它们的实际功能。

综上所述，合理地控制并发请求可以有效地提高系统的稳定性和性能，需要根据系统的实际情况选择合适的并发控制方法，并结合监控和调优来持续优化系统的性能。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/617691
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号