线程：搜索一个值并停止所有线程

Question

我需要实现一个搜索方法，该方法将在 haystack 中搜索并返回第一个建立的 needle 索引。

静态整数搜索(T needle, T[] haystack, int numThreads)

我的问题：如果其中一个线程找到结果，我该如何停止所有线程？
例如：我正在搜索 5，我在数组中有 10 个数字，这样 [2, 4, 5, 6, 1, 4, 5, 8, 9, 3] 并且有 2 个线程。所以第一个线程将寻找第一部分 [0 - 5)，第二个线程将搜索其他部分 [5 - 10)。如果线程2首先启动并且比其他线程更快地找到结果，它应该返回6并终止线程1和2。

Answer 1

执行此操作的经典方法是简单地在线程之间共享数据，以便它们可以相互通信。换句话说，在启动线程之前将一些标志值初始化为“未找到”。

然后，当线程运行时，它们会处理数组中的元素，直到它们的元素用完，或标志值已设置为“找到”。

在伪代码中，应该是这样的：

main():
    global array = createArray(size = 10000, values = random)
    global foundIndex = -1
    global mutex = createMutex()

    startThread(id = 1, func = threadFn, param = (0, 4999))
    startThread(id = 2, func = threadFn, param = (5000, 9999))

    waitFor(id = 1)
    waitFor(id = 2)

    print("Result is ", foundIndex)

threadFn(first, last):
    for index in first through last inclusive:
        if userSpecifiedCheckFound(array[index]):
            mutex.lock()
            if foundIndex == -1:
                foundIndex = index
            mutex.unlock()
            return

        mutex.lock()
        localIndex = foundIndex
        mutex.unlock()
        if localIndex != -1:
            return

您可以从中看出，该函数的每个实例都会设置共享数据，并在找到与您要查找的任何条件匹配的值时返回。如果另一个线程已经设置了共享数据，它也会返回（不设置共享数据），这意味着如果另一个线程已经找到了某些东西，它可以提前退出。

请记住，在这种情况下，共享数据 foundIndex 需要受到保护，以免同时发生更改，以免损坏。在伪代码中，我展示了如何使用低级互斥信号量来做到这一点。

---

在 Java 中，这意味着使用synchronized 来实现相同的效果。例如，下面的代码设置了一些合适的测试数据，这样 20 单元数组中的第 16 单元将满足搜索条件。

然后它运行两个线程，一个处理每一半数据，直到找到那个单元格。

public class TestProg extends Thread {
  // Shared data.

  static int [] sm_array = new int[20];
  static int sm_foundIndex = -1;

  // Each thread responsible for its own stuff.

  private int m_id, m_curr, m_last;
  public TestProg(int id, int first, int last) {
    m_id = id;
    m_curr = first;
    m_last = last;
  }

  // Runnable: continue until someone finds it.

  public void run() {
    // Try all cells allotted to thread.

    while (m_curr <= m_last) {
      System.out.println(m_id + ": processing " + m_curr);

      // If I find it first, save and exit.

      if (sm_array[m_curr] != 0) {
        synchronized(this) {
          if (sm_foundIndex == -1) {
            sm_foundIndex = m_curr;
            System.out.println(m_id + ": early exit, I found it");
            return;
          }
        }
      }

      // If someone else finds it, just exit.

      synchronized(this) {
        if (sm_foundIndex != -1) {
          System.out.println(m_id + ": early exit, sibling found it");
          return;
        }
      }

      // Kludge to ensure threads run side-by-side.

      try { Thread.sleep(100); } catch(Exception e) {}

      m_curr++;
    }
  }

  public static void main(String[] args) {
    // Create test data.

    for (int i = 0; i < 20; i++) {
      sm_array[i] = 0;
    }
    sm_array[15] = 1;

    // Create and start threads.

    HelloWorld thread1 = new HelloWorld(1, 0, 9);
    HelloWorld thread2 = new HelloWorld(2, 10, 19);
    thread1.start();
    thread2.start();

    // Wait for both to finish, then print result.

    try {
      thread1.join();
      thread2.join();
      System.out.println("=> Result was " + sm_foundIndex);
    } catch(Exception e) {
      System.out.println("Interrupted: " + e);
    }
  }
}

该代码的输出（尽管线程使它有点不确定）是：

1: processing 0
2: processing 10
1: processing 1
2: processing 11
1: processing 2
2: processing 12
1: processing 3
2: processing 13
1: processing 4
2: processing 14
1: processing 5
2: processing 15
2: early exit, I found it
1: processing 6
1: early exit, sibling found it
=> Result was 15

Answer 2

您可以查看ExecutorCompletionService，一旦第一个结果可用，然后取消所有其他任务。

CompletionService 使用提供的Executor 来执行任务和 将所有未来的结果放在一个队列中，您可以按照完成的顺序从中获取结果