排序线程按照它们创建/启动的顺序运行

Question

我如何按照线程实例化的顺序对线程进行排序。例如，如何使下面的程序按顺序打印数字 1...10。

public class ThreadOrdering {
    public static void main(String[] args) {

        class MyRunnable implements Runnable{
            private final int threadnumber;

            MyRunnable(int threadnumber){
                this.threadnumber = threadnumber;
            }

            public void run() {
                System.out.println(threadnumber);
            }
        }

        for(int i=1; i<=10; i++){
            new Thread(new MyRunnable(i)).start();
        }
    }
}

Answer 1

听起来你想要 ExecutorService.invokeAll，它将以固定顺序返回工作线程的结果，即使它们可能以任意顺序调度：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ThreadOrdering {

    static int NUM_THREADS = 10;

    public static void main(String[] args) {
        ExecutorService exec = Executors.newFixedThreadPool(NUM_THREADS);
        class MyCallable implements Callable<Integer> {
            private final int threadnumber;

            MyCallable(int threadnumber){
                this.threadnumber = threadnumber;
            }

            public Integer call() {
                System.out.println("Running thread #" + threadnumber);
                return threadnumber;
            }
        }

        List<Callable<Integer>> callables =
            new ArrayList<Callable<Integer>>();
        for(int i=1; i<=NUM_THREADS; i++) {
            callables.add(new MyCallable(i));
        }
        try {
            List<Future<Integer>> results =
                exec.invokeAll(callables);
            for(Future<Integer> result: results) {
                System.out.println("Got result of thread #" + result.get());
            }
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        } catch (ExecutionException ex) {
            ex.printStackTrace();
        } finally {
            exec.shutdownNow();
        }
    }

}

Answer 2

“我实际上有一些部分想要并行执行，然后一旦生成结果，我想按特定顺序合并结果。”谢谢，这澄清了你的问题。

您可以同时运行它们，但重要的是当线程完成计算时按顺序获取它们的结果。 Thread#join() 按照您想要获取结果的顺序排列它们，或者只是 Thread#join() 它们全部然后迭代它们以获得结果。

// Joins the threads back to the main thread in the order we want their results.
public class ThreadOrdering {
    private class MyWorker extends Thread {
        final int input;
        int result;
        MyWorker(final int input) {
            this.input = input;
        }
        @Override
        public void run() {
            this.result = input; // Or some other computation.
        }
        int getResult() { return result; }
    }

    public static void main(String[] args) throws InterruptedException {
        MyWorker[] workers = new MyWorker[10];
        for(int i=1; i<=10; i++) {
            workers[i] = new MyWorker(i);
            workers[i].start();
        }

        // Assume it may take a while to do the real computation in the threads.

        for (MyWorker worker : workers) {
            // This can throw InterruptedException, but we're just passing that.
            worker.join();
            System.out.println(worker.getResult());
        }
    }
}

Answer 3

简单来说，线程的调度是不确定的。

http://www.janeg.ca/scjp/threads/scheduling.html 死域 - 请勿点击

上述页面的 WaybackMachine 版本

较长的答案是如果您想执行此操作，则需要手动等待每个线程完成其工作，然后才能允许另一个线程运行。请注意，以这种方式，所有线程仍将交错，但在您允许之前它们不会完成任何工作。看一下同步保留字。

Answer 4

您可以将它们链接起来 - 也就是说，让第一个启动第二个，第二个启动第三个，等等。它们不会真正同时运行，除了每个启动时有一点重叠之外。

public class ThreadOrdering
{
    public static void main(String[] args)
    {
        MyRunnable[] threads = new MyRunnable[10];//index 0 represents thread 1;
        for(int i=1; i<=10; i++)
            threads[i] = new MyRunnable(i, threads); 
        new Thread(threads[0].start);  
    }
}

public class MyRunnable extends Runnable
{
    int threadNumber;
    MyRunnable[] threads;

    public MyRunnable(int threadNumber, MyRunnable[] threads)
    {
        this.threadnumber = threadnumber;
        this.threads = threads;
    }

    public void run()
    {
        System.out.println(threadnumber);
        if(threadnumber!=10)
            new Thread(threadnumber).start();
    }
}

Answer 5

这是一种无需等待每个结果的主线程即可完成此操作的方法。相反，让工作线程共享一个对结果进行排序的对象。

import java.util.*;

public class OrderThreads {
    public static void main(String... args) {
        Results results = new Results();
        new Thread(new Task(0, "red", results)).start();
        new Thread(new Task(1, "orange", results)).start();
        new Thread(new Task(2, "yellow", results)).start();
        new Thread(new Task(3, "green", results)).start();
        new Thread(new Task(4, "blue", results)).start();
        new Thread(new Task(5, "indigo", results)).start();
        new Thread(new Task(6, "violet", results)).start();
    }
}

class Results {
    private List<String> results = new ArrayList<String>();
    private int i = 0;

    public synchronized void submit(int order, String result) {
        while (results.size() <= order) results.add(null);
        results.set(order, result);
        while ((i < results.size()) && (results.get(i) != null)) {
            System.out.println("result delivered: " + i + " " + results.get(i));
            ++i;
        }
    }
}


class Task implements Runnable {
    private final int order;
    private final String result;
    private final Results results;

    public Task(int order, String result, Results results) {
        this.order = order;
        this.result = result;
        this.results = results;
    }

    public void run() {
        try {
            Thread.sleep(Math.abs(result.hashCode() % 1000)); // simulate a long-running computation
        }
        catch (InterruptedException e) {} // you'd want to think about what to do if interrupted
        System.out.println("task finished: " + order + " " + result);
        results.submit(order, result);
    }
}

Answer 6

如果您需要细粒度的控制，则不应使用线程，而应考虑将合适的 Executor 与 Callables 或 Runnables 一起使用。请参阅 Executors 类以获取广泛的选择。

Answer 7

一个简单的解决方案是使用锁数组A（每个线程一个锁）。当线程i 开始执行时，它会获取关联的锁A[i]。当它准备好合并其结果时，它会释放其锁 A[i] 并等待锁 A[0]、A[1]、...、A[i - 1]待发布；然后它合并结果。

（在这种情况下，线程i表示第i启动的线程）

我不知道在Java中使用什么类，但它必须很容易实现。您可以开始阅读此。

如果您还有更多问题，请随时提问。

Answer 8

public static void main(String[] args) throws InterruptedException{
    MyRunnable r = new MyRunnable();
    Thread t1 = new Thread(r,"A");
    Thread t2 = new Thread(r,"B");
    Thread t3 = new Thread(r,"C");

    t1.start();
    Thread.sleep(1000);

    t2.start();
    Thread.sleep(1000);
    t3.start();
}

Answer 9

使用信号量可以很容易地实现线程执行顺序的控制。所附代码基于 Schildt 关于 Java 的书（完整参考......）中提出的想法。
// 基于以下内容中提出的想法：
// Schildt H.：Java.The.Complete.Reference.9th.Edition。

import java.util.concurrent.Semaphore;

class Manager {
    int n;
// Initially red on semaphores 2&3; green semaphore 1.
    static Semaphore SemFirst = new Semaphore(1);
    static Semaphore SemSecond = new Semaphore(0);
    static Semaphore SemThird = new Semaphore(0);

void firstAction () {
    try {
        SemFirst.acquire();
    } catch(InterruptedException e) {
        System.out.println("Exception InterruptedException catched");
    }
    System.out.println("First: " );
    System.out.println("-----> 111");
    SemSecond.release();
}
void secondAction() {
    try{
        SemSecond.acquire();
    } catch(InterruptedException e) {
        System.out.println("Exception InterruptedException catched");
    }
    System.out.println("Second: ");
    System.out.println("-----> 222");
    SemThird.release();
}
void thirdAction() {
    try{
        SemThird.acquire();
    } catch(InterruptedException e) {
        System.out.println("Exception InterruptedException catched");
    }
    System.out.println("Third: ");
    System.out.println("-----> 333");
    SemFirst.release();
}
}

class Thread1 implements Runnable {
    Manager q;

    Thread1(Manager q) {
    this.q = q;
    new Thread(this, "Thread1").start();
}

public void run() {
    q.firstAction();
}
}

class Thread2 implements Runnable {
    Manager q;

    Thread2(Manager q) {
    this.q = q;
    new Thread(this, "Thread2").start();
}

public void run() {
    q.secondAction();
}
}

class Thread3 implements Runnable {
    Manager q;

    Thread3(Manager q) {
    this.q = q;
    new Thread(this, "Thread3").start();
}

public void run() {
    q.thirdAction();
}
}

class ThreadOrder {
    public static void main(String args[]) {
    Manager q = new Manager();
    new Thread3(q);
    new Thread2(q);
    new Thread1(q);
    }
}

Answer 10

这可以不使用synchronized关键字并在易失性关键字的帮助下完成。以下是代码。

package threadOrderingVolatile;

public class Solution {
    static volatile int counter = 0;
    static int print = 1;
    static char c = 'A';

    public static void main(String[] args) {
        // TODO Auto-generated method stub
        Thread[] ths = new Thread[4];
        for (int i = 0; i < ths.length; i++) {
            ths[i] = new Thread(new MyRunnable(i, ths.length));
            ths[i].start();
        }
    }

    static class MyRunnable implements Runnable {
        final int thID;
        final int total;

        public MyRunnable(int id, int total) {
            thID = id;
            this.total = total;
        }

        @Override
        public void run() {
            while(true) {
                if (thID == (counter%total)) {
                    System.out.println("thread " + thID + " prints " + c);
                    if(c=='Z'){
                        c='A';
                    }else{
                        c=(char)((int)c+1);
                    }
                    System.out.println("thread " + thID + " prints " + print++);
                    counter++;                  
                } else {
                    try {
                        Thread.sleep(30);
                    } catch (InterruptedException e) {
                        // log it
                    }
                }
            }
        }

    }

}

以下是 github 链接，其中有一个自述文件，其中详细解释了它是如何发生的。

Answer 11

如果有人需要，这是我在 Python 中解决这个问题的看法。我的方法是使用双端队列。该类的每个实例都将自己放入队列中，在构造时执行工作任务，然后等待队列的前一个成员终止其输出以完成其任务。诀窍是每个实例都有两个锁（也称为信号量）来控制打印。从队列中输入的最后一个实例复制的一个：我将等待此锁打印。另一种是在构造时创建并锁定的。进入队列的下一个实例将复制该实例，并且必须等待我释放它，我将在打印完成后执行此操作。

import random
from functools import \
    partial  # an object made of a function ready to call with prefilled arguments
import threading, time
from collections import deque

printing_lock = threading.Lock()  # for debuging: avoid threads printing debug infos in the middle of each other


class WorkNowPrintLater:
    def __init__(self, id: int, work_func: partial, print_func: partial, my_queue: deque, queue_lock: threading.Lock):
        """
        Create an instance of this class does the whole thing:
            1/ do some job immediately
            2/ do some other job (usually an output) in the same order as instances were created
        :param id:
            id of this instance. for debug purpose only
        :param work_func:
            the working function.
            Note for non python users: a partial is an object made of a function with its arguments
            already in place, ready to launch
        :param print_func:
            the output function
        :param my_queue:
            a double-end-queue (deque) where each instance of WorkNowPrintLater will put itself on construction
            Must be the same for all instances to synchronize
        :param queue_lock:
            a semaphore to protect access to the queue
            Must be the same for all instances to synchronize
        """
        with printing_lock:
            print(f"{id} being created")
        self.id = id
        self.work_func = work_func
        self.print_func = print_func
        self.my_queue = my_queue
        self.queue_lock = queue_lock
        self.done = False
        self.lock_i_wait_before_i_printing = threading.Lock() if len(my_queue) == 0 else my_queue[
            -1].lock_you_wait_before_you_print
        self.lock_you_wait_before_you_print = threading.Lock()
        self.lock_you_wait_before_you_print.acquire()  # Block access to printer to the next element in the queue until I'm done

        with printing_lock:
            print(f"{self.id} waits for queue lock to put myself in")
        with self.queue_lock:
            self.my_queue.append(self)

        my_worker = threading.Thread(target=self.worker)
        my_worker.start()
        print(f"{id} is alive")

    def worker(self):
        with printing_lock:
            print(f"{self.id} working starts")
        self.work_func()

        with printing_lock:
            print(f"{self.id} working done. wait for printer")
        with self.lock_i_wait_before_i_printing:
            self.print_func()
            self.lock_you_wait_before_you_print.release()
            self.done = True

        with printing_lock:
            print(f"{self.id} waits for queue lock to do housekeeping")
        with self.queue_lock:
            with printing_lock:
                print(f"{self.id} housekeeping starts")
            if len(self.my_queue) <= 1:
                return
            while self.my_queue[0].done and len(self.my_queue) >= 2:
                _ = self.my_queue.popleft()


 """ MAIN PROGRAM """


threads_queue = deque()
semaphore = threading.Lock()


def wait(id):
    t = random.random() * 10
    print(f"{id} is working for {t} seconds")
    time.sleep(t)


def output(id):
    print(f"output {id}")


for i in range(5):
    _ = WorkNowPrintLater(i, partial(wait, i), partial(output, i), threads_queue, semaphore)