java中的最小时间锁

Question

我正在寻找一种方法来限制对 java 中方法的访问不超过每 X 秒一次。这是我的情况：

我想用多个线程并行运行这段代码：

private MyService service;
public void run() {
    // Send request to remote service
    InputStream response = service.executeRequest();

    // Process response
    ... some more code
}

executeRequest() 方法向远程服务器发送一个 http 请求（这不是我的，我无权访问它的实现）并等待来自服务器的响应。然后它对数据进行一些处理。 我想让许多线程并行运行。我的问题是如果同时发送太多请求，远程服务器会崩溃。所以我想要一些方法来确保 executeRequest() 方法每秒不会被调用超过一次。

你知道我如何在 java 中做到这一点吗？谢谢

Answer 1

您是否考虑过使用sleep 让线程在跳转到远程调用之前暂停？您可以让线程在 1 到 5 之间随机休眠 # 秒，这将限制任何时候访问该方法的线程数。

您还可以在 1 秒后过期的方法上放置 lock，以便每个线程“获取”锁，执行该方法，但它的锁过期，因此下一个线程可以获取并执行。将锁放在方法的开头——让它什么都不做，除了保持线程一秒钟thread.sleep(1000)，然后继续执行。这将限制您一次只有一个线程访问该方法。

编辑：对下面 OP 的评论的回应

   class X {
   private final ReentrantLock lock = new ReentrantLock();
   // ...

   public void m() { 
     lock.lock();  // block until condition holds
     try {
       thread.sleep(1000) //added to example by floppydisk.
     } finally {
       lock.unlock()
       doYourConnectionHere();
     }
   }
 }

修改自：ReentrantLock。在 try/catch 中，你所做的只是 thread.sleep(1000) 而不是实际做某事。然后它释放下一个线程的锁并继续执行方法体的其余部分——在你的例子中是到远程服务器的连接。

Answer 2

Hrm，我不确定限制访问方法的频率是否会防止过载。

也许上面的帖子中没有足够的信息，但似乎 WorkerThread + JobQueue 设置在这里可以正常工作。

深思： Multithreaded job queue manager

编辑：尽量不那么含糊……

• 让服务器将请求收集到某种数据结构中， 也许是一个叫 Job 的类。
• 然后将作业放置到底部 一个队列。
• 让 WorkerThread 对象弹出 Job 对象 队列顶部并处理它们。
• 确保只实例化为 许多 WorkerThread 对象，因为您需要维持适当的服务器负载。 只有实验才能确定这个数字，但作为一个非常粗略的 规则，从 # of processing cores - 1 开始。（又名开始 7 个工作人员 8核机器）

EDIT #2根据新信息：

• 在客户端设置队列
• 创建一个 Worker，可以跟踪哪些作业已提交、哪些作业已得到响应以及哪些作业仍在处理中。这将允许限制在任何时间提交的作业数量。
• 让 Worker 跟踪“lastSubmissionTime”以防止任何提交发生在距上一秒不到 1 秒

Answer 3

在调用worker执行服务的控制器类中，使用ExecutorService启动线程池以利用worker类。

ExecutorService pool = Executors.newFixedThreadPool(10);
pool.submit(new MyServiceImpl(someObject));

要添加其他人已经建议的内容，请让工作人员类从队列中执行任务并等待所需的分钟数，然后再从队列中获取另一个任务。我以这个 2min 为例。

例子：

public class MyServiceImpl implements MyService , Runnable {

  public static final int MAX_SIZE = 10;
  private final BlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(MAX_SIZE);

    @Override
    public void run() {
    try
    {
     Object obj;
      while ((obj==queue.take()) != null)
      {
       executeRequest(obj);
       //wait for 2 min
       Thread.sleep(1000 * 60 * 2);
      }
    }
    catch (InterruptedException e)
    {}
  }

  public void executeRequest(Object obj)
  {
    // do routine
  }  

  public MyServiceImpl (Object token)
  {
    try
    {
      queue.put(token);
    }
    catch (InterruptedException e)
    {
      throw new AssertionError(e);
    }
  }
}

Answer 4

在客户端限制并发不是一个好的模式——客户端应该如何相互了解？

Answer 5

您可以使用信号量来限制能够调用 executeRequest() 的线程数：

http://download.oracle.com/javase/1,5,0/docs/api/java/util/concurrent/Semaphore.html

执行线程可以在进入执行之前增加信号量，其他线程可以等待它下降到 0 或反映允许并行运行的数量的数字。

定时任务：

http://download.oracle.com/javase/1.4.2/docs/api/java/util/TimerTask.html

可用于在 3 秒后减少信号量...将条目限制为每 3 秒不超过 1 个新进入者：

Answer 6

使用动态代理，您可以包装您的服务并在 InvocationHandler 中处理最大执行：

MyService proxy = (MyService) Proxy.newProxyInstance( //
    MyService.class.getClassLoader(), //
    new Class[] {MyService.class}, //
    new MaxInvocationHandler());

InvocationHandler 的幼稚实现可能如下所示：

  class MaxInvocationHandler implements InvocationHandler {
    private static final long MAX_INTERVAL = 1000L;
    private static final long MAX_INVOCATIONS = 1;

    AtomicLong time = new AtomicLong();
    AtomicLong counter = new AtomicLong();

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
      long currentTime = System.currentTimeMillis();
      if (time.get() < currentTime) {
        time.set(currentTime + MAX_INTERVAL);
        counter.set(1);
      } else if(counter.incrementAndGet() > MAX_INVOCATIONS) {
        throw new RuntimeException("Max invocation exceeded");
      }

      return method.invoke(proxy, args);
    }
  }