Java

Question

我有给定随机数（1 到 n）的线程，并被指示按排序顺序打印它们。我使用了信号量，这样我就获得了许可数 = 随机数，并比获得的多释放了一个许可。

acquired = random number; released = 1+random number

信号量的初始许可计数为 1。因此随机数为 1 的线程应该获得许可，然后是 2，依此类推。

根据下面给出的文档，这是受支持的

There is no requirement that a thread that releases a permit must have acquired that permit by calling acquire().

问题是我的程序在 n>2 的 1 之后卡住了。

我的程序如下：

import java.util.concurrent.Semaphore;

public class MultiThreading {
    public static void main(String[] args) {
        Semaphore sem = new Semaphore(1,false);
        for(int i=5;i>=1;i--)
            new MyThread(i, sem);
    }
}
class MyThread implements Runnable {
    int var;Semaphore sem;
    public MyThread(int a, Semaphore s) {
        var =a;sem=s;
        new Thread(this).start();
    }
    @Override
    public void run() {
        System.out.println("Acquiring lock -- "+var);
        try {
            sem.acquire(var);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(var);

        System.out.println("Releasing lock -- "+var);
        sem.release(var+1);
    }
}

输出为：

Acquiring lock -- 4
Acquiring lock -- 5
Acquiring lock -- 3
Acquiring lock -- 2
Acquiring lock -- 1
1
Releasing lock -- 1

虽然如果我使用 tryAcquire 修改我的代码，它运行非常好。下面是新的运行实现

@Override
public void run() {
    boolean acquired = false;
    while(!acquired) {
        acquired = sem.tryAcquire(var);
    }
    System.out.println(var);
    sem.release(var+1);
}

有人可以解释当多个线程等待不同的许可请求时信号量的许可获取机制吗？

Answer 1

Semaphore.acquire(int) 的 Javadoc 说：

If insufficient permits are available then the current thread becomes 
disabled for thread scheduling purposes and lies dormant until one of 
two things happens:

Some other thread invokes one of the release methods for this semaphore, 
the current thread is next to be assigned permits and the number of 
available permits satisfies this request [or the thread is interrupted].

在您的示例中，"next to be assigned" 的线程可能是线程 4。它正在等待直到有 4 个可用的许可证。但是，调用 acquire() 时获得许可的线程 1 仅释放了 2 个许可，这不足以解除线程 4 的阻塞。同时，线程 2 是唯一有足够许可的线程，它不是下一个待分配，因此未获得许可。

您修改后的代码运行良好，因为线程在尝试获取信号量时不会阻塞；他们只是再试一次，走到队伍的后面。最终线程 2 到达了队列的最前面，因此是下一个被分配的对象，因此获得了它的许可。

Answer 2

这是一个聪明的策略，但您误解了 Sempahore 如何发放许可证。如果您运行您的代码足够多次，您实际上会看到它到达第二步：

Acquiring lock -- 5
Acquiring lock -- 1
1
Releasing lock -- 1
Acquiring lock -- 3
Acquiring lock -- 2
2
Acquiring lock -- 4
Releasing lock -- 2

如果您继续重新运行足够多次，您实际上会看到它成功完成。发生这种情况是因为 Semaphore 分发许可证的方式。您假设 Semaphore 将在获得足够的许可后立即尝试容纳 acquire() 调用。如果我们仔细查看 Semaphore.aquire(int) 的文档，我们会发现情况并非如此（强调我的）：

If insufficient permits are available then the current thread becomes disabled for thread scheduling purposes and lies dormant until ... some other thread invokes one of the release methods for this semaphore, the current thread is next to be assigned permits and the number of available permits satisfies this request.

换句话说，Semaphore 保留一个等待获取请求的队列，并且在每次调用 .release() 时， 只检查队列的头部 。特别是如果您启用公平排队（将第二个构造函数参数设置为 true），您会看到甚至不会发生第一步，因为第 5 步（通常）是队列中的第一个，甚至是新的 acquire() 可以完成的呼叫将排在其他未决呼叫之后。

简而言之，这意味着您不能像代码假设的那样尽快依赖 .acquire() 到 return。

通过在循环中使用 .tryAcquire()，您可以避免进行任何阻塞调用（因此会给您的 Semaphore 带来更多的负载），并且一旦必要数量的许可可用tryAcquire() 调用将成功获取它们。这可行但很浪费。

想象一下餐厅的候补名单。使用 .aquire() 就像将您的名字放在列表中并等待被调用。它可能不是非常有效，但他们会在（合理的）相当长的时间内找到你。相反，想象一下，如果每个人都尽可能多地对主持人 "Do you have a table for n yet?" 大喊大叫 - 这就是您的 tryAquire() 循环。它可能仍然有效（就像在您的示例中所做的那样），但这肯定不是正确的方法。

那么你应该怎么做呢？ java.util.concurrent 中有许多可能有用的工具，哪种工具最好取决于您要做什么。看到你有效地让每个线程启动下一个线程，我可能会使用 BlockingQueue 作为同步辅助，每次都将下一步推入队列。然后每个线程都会轮询队列，如果没有轮到激活的线程，则替换该值并再次等待。

这是一个例子：

public class MultiThreading {
  public static void main(String[] args) throws Exception{
    // Use fair queuing to prevent an out-of-order task
    // from jumping to the head of the line again
    // try setting this to false - you'll see far more re-queuing calls
    BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(1, true);
    for (int i = 5; i >= 1; i--) {
      Thread.sleep(100); // not necessary, just helps demonstrate the queuing behavior
      new MyThread(i, queue).start();
    }
    queue.add(1); // work starts now
  }

  static class MyThread extends Thread {
    int var;
    BlockingQueue<Integer> queue;

    public MyThread(int var, BlockingQueue<Integer> queue) {
      this.var = var;
      this.queue = queue;
    }

    @Override
    public void run() {
      System.out.println("Task " + var + " is now pending...");
      try {
        while (true) {
          int task = queue.take();
          if (task != var) {
            System.out.println(
                "Task " + var + " got task " + task + " instead - re-queuing");
            queue.add(task);
          } else {
            break;
          }
        }
      } catch (InterruptedException e) {
        // If a thread is interrupted, re-mark the thread interrupted and terminate
        Thread.currentThread().interrupt();
        return;
      }

      System.out.println("Finished task " + var);

      System.out.println("Registering task " + (var + 1) + " to run next");
      queue.add(var + 1);
    }
  }
}

这将打印以下内容并成功终止：

Task 5 is now pending...
Task 4 is now pending...
Task 3 is now pending...
Task 2 is now pending...
Task 1 is now pending...
Task 5 got task 1 instead - re-queuing
Task 4 got task 1 instead - re-queuing
Task 3 got task 1 instead - re-queuing
Task 2 got task 1 instead - re-queuing
Finished task 1
Registering task 2 to run next
Task 5 got task 2 instead - re-queuing
Task 4 got task 2 instead - re-queuing
Task 3 got task 2 instead - re-queuing
Finished task 2
Registering task 3 to run next
Task 5 got task 3 instead - re-queuing
Task 4 got task 3 instead - re-queuing
Finished task 3
Registering task 4 to run next
Task 5 got task 4 instead - re-queuing
Finished task 4
Registering task 5 to run next
Finished task 5
Registering task 6 to run next

Java - 不获取信号量释放

Java - Semaphore release without acquire

concurrency

multithreading

semaphore