2

在我编写的 Web 服务器中，每个请求都会调用一个操作列表。其中一些操作不像其他操作那么重要，所以我想在后台线程中运行它们。

此外，由于它们并不那么重要，我不在乎它们中的一个是否很少失败，并且我不希望它们永远占用一个线程，因此其他线程将可用于处理下一批。

所以，我想要一个线程池（例如：10个线程）并像这样为每个后台任务分配一个线程。将每个线程限制为 1 秒，如果到此时间还没有完成，则将其杀死，以便下一个任务进入。

我该怎么做呢？

到目前为止，这就是我所拥有的：

public class AsyncCodeRunner {

    private static final ExecutorService executor = Executors.newFixedThreadPool(10);

    public void Run(Callable<Void> callableCode, int timeout) {

        final int threadTimeout = 10;
        Future<Void> callableFuture = executor.submit(callableCode);

        try {
            callableFuture.get(threadTimeout, TimeUnit.SECONDS);
        } catch (Exception e) {
            logger.Info("Thread was timed out", e);
        }
    }
}

我想像这样使用这个类：

public void processRequest(RequestObject request) {

    // do some important processing

    // throw some less important processing to background thread
    (new AsyncCodeRunner()).Run(new Callable<Void> () {
        @Override
        public Void call() throws Exception {
            // do something...
            return null;
        }
    }, 1); // 1 second timeout

    // return result (without waiting for background task)
    return;

}

这会像我想要的那样工作吗？或者我应该如何改变它呢？
如果我打电话Run()但线程池中没有可用的线程可以分发怎么办？

4

2 回答 2

2

我认为你对这个相当优雅的想法的主要问题是你只是超时了get，Future一旦超时，你实际上并没有中止这个过程，你只是放弃等待它。当您意识到您甚至可能在进程尚未开始时超时时，问题变得更加复杂 - 它仍然在队列中。

也许这样的事情会很有效。它确实需要两个线程，但一个TimerTask线程应该消耗很少。

public class RunWithTimeout {

    public RunWithTimeout(Runnable r, long timeout) {
        // Prepare the thread.
        final Thread t = new Thread(r);
        // Start the timer.
        new Timer(true).schedule(new TimerTask() {

            @Override
            public void run() {
                if (t.isAlive()) {
                    // Abort the thread.
                    t.interrupt();
                }
            }
        }, timeout * 1000);
        // Start the thread.
        t.start();
    }

}

class WaitAFewSeconds implements Runnable {

    final long seconds;

    WaitAFewSeconds(long seconds) {
        this.seconds = seconds;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(seconds * 1000);
        } catch (InterruptedException ie) {
            System.out.println("WaitAFewSeconds(" + seconds + ") - Interrupted!");
        }
    }

}

public void test() {
    new RunWithTimeout(new WaitAFewSeconds(5), 3);
    new RunWithTimeout(new WaitAFewSeconds(3), 5);
}

这是一种仅使用一个额外线程的替代方案。

public class ThreadKiller implements Runnable {

    DelayQueue<WaitForDeath> kill = new DelayQueue<>();

    private class WaitForDeath implements Delayed {

        final Thread t;
        final long finish;

        public WaitForDeath(Thread t, long wait) {
            this.t = t;
            this.finish = System.currentTimeMillis() + wait;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(finish - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            long itsFinish = ((WaitForDeath) o).finish;
            return finish < itsFinish ? -1 : finish == itsFinish ? 0 : 1;
        }

    }

    @Override
    public void run() {
        while (true) {
            try {
                WaitForDeath t = kill.take();
                if (t.t.isAlive()) {
                    // Interrupt it.
                    t.t.interrupt();
                }
            } catch (InterruptedException ex) {
                // Not sure what to do here.
            }
        }
    }

    public void registerThread(Thread t, long wait) {
        // Post it into the delay queue.
        kill.add(new WaitForDeath(t, wait));
    }
}

public void test() throws InterruptedException {
    // Testing the ThreadKiller.
    ThreadKiller killer = new ThreadKiller();
    Thread killerThread = new Thread(killer);
    killerThread.setDaemon(true);
    Thread twoSeconds = new Thread(new WaitAFewSeconds(2));
    Thread fourSeconds = new Thread(new WaitAFewSeconds(4));
    killer.registerThread(twoSeconds, 5000);
    killer.registerThread(fourSeconds, 3000);
    killerThread.start();
    twoSeconds.start();
    fourSeconds.start();
    System.out.println("Waiting");
    Thread.sleep(10 * 1000);
    System.out.println("Finished");
    killerThread.interrupt();
}
于 2015-01-23T10:51:23.327 回答
0

您需要在线程运行时启动计时器。然后不会杀死处于等待状态的线程。这是该线程的示例：

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class PoolTest {
    class TimeOutTask extends TimerTask {
        Thread t;

        TimeOutTask(Thread t) {
            this.t = t;
        }

        public void run() {
            if (t != null && t.isAlive()) {
                t.interrupt();
            }
        }
    }

    class MyRunnable implements Runnable {
        Timer timer = new Timer(true);

        public void run() {
            timer.schedule(new TimeOutTask(Thread.currentThread()), 1000);
            try {
                System.out.println("MyRunnable...");
                Thread.sleep(10000);
            } catch (InterruptedException ie) {
                System.out.println("MyRunnable error...");
                ie.printStackTrace();
            }
        }
    }

    public static void main(String args[]) {
        new PoolTest();
    }

    public PoolTest() {
        try {
            ExecutorService pe = Executors.newFixedThreadPool(3);
            pe.execute(new MyRunnable());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
于 2019-03-06T12:56:06.730 回答