ExecutorService固定线程拒绝

Question

我有一些线程的问题。

我的剧本

1-从文本文件将1000多万行加载到数组中

2-创建5个固定线程的执行池

3-然后它正在迭代该列表并将一些线程添加到队列中

executor.submit(new MyCustomThread(line,threadTimeout,"[THREAD "+Integer.toString(increment)+"]"));

现在活动线程永远不会绕过5个固定线程，这很好，但我发现我的处理器进入100%负载，我已经调试了一点，我看到正在调用MyCustomThread构造函数，女巫意味着无论我声明5个固定线程，ExecutorService仍将尝试创建10万个对象。

主要问题是：我如何防止这种情况？我只是想让线程在没有空间的情况下被拒绝，而不是创建1000万对象并一个接一个地运行它们。

第二个问题：我如何得到当前的活动线程？我尝试了threadGroup.activeCount（），但它总是给我5 5 5......

调用方类：

System.out.println("Starting threads ...");
final ThreadGroup threadGroup = new ThreadGroup("workers");
//ExecutorService executor = Executors.newFixedThreadPool(howManyThreads);

ExecutorService executor = Executors.newFixedThreadPool(5,new ThreadFactory() {
    public Thread newThread(Runnable r) {
        return new Thread(threadGroup, r);
    }
});

int increment = 0;              
for(String line : arrayOfLines)
{
    if(increment > 10000)
    {
        //System.out.println("TOO MANY!!");
        //System.exit(0);
    }

    System.out.println(line);
    System.out.println(threadGroup.activeCount());

    if(threadGroup.activeCount() >= 5)
    {
        for(int i = 0; i < 10; i++)
        {
            System.out.println(threadGroup.activeCount());
            System.out.println(threadGroup.activeGroupCount());
            Thread.sleep(1000);
        }
    }


    try
    {
        executor.submit(new MyCustomThread(line,threadTimeout,"[THREAD "+Integer.toString(increment)+"]"));
    }
    catch(Exception ex)
    {
        continue;
        //System.exit(0);
    }

    increment++;
}

executor.awaitTermination(10, TimeUnit.MILLISECONDS);
executor.shutdown();

Thread等级：

public class MyCustomThread extends Thread
{
    private String ip;
    private String threadName;
    private int threadTimeout = 10;

    public MyCustomThread(String ip)
    {
        this.ip = ip;
    }

    public MyCustomThread(String ip,int threadTimeout,String threadName)
    {

        this.ip = ip;
        this.threadTimeout = threadTimeout;
        this.threadName = threadName;

        System.out.prinln("MyCustomThread constructor has been called!");
    }

    @Override
    public void run()
    {
        // do some stuff that takes time ....
    }
}

谢谢你。

Answer 1

我认为这里最大的问题是MyCustomThread应该实现Runnable，而不是扩展Thread。使用ExecutorService时，您可以让它处理线程管理（即不需要创建它们）

这是我认为你想要做的一个近似值。希望这有帮助。

public class FileProcessor
{

    public static void main(String[] args)
    {

        List<String> lines = readFile();
        System.out.println("Starting threads ...");
        ExecutorService executor = Executors.newFixedThreadPool(5);

        for(String line : lines)
        {
            try
            {
                executor.submit(new MyCustomThread(line));
            }
            catch(Exception ex)
            {
                ex.printStackTrace();
            }
        }

        try
        {
            executor.shutdown();
            executor.awaitTermination(10, TimeUnit.SECONDS);
        }
        catch (InterruptedException e)
        {
            System.out.println("A processor took longer than the await time to complete.");
        }
        executor.shutdownNow();

    }

    protected static List<String> readFile()
    {
        List<String> lines = new ArrayList<String>();
        try
        {
            String filename = "/temp/data.dat";
            FileReader fileReader = new FileReader(filename );
            BufferedReader bufferedReader = new BufferedReader(fileReader);
            String line = null;
            while ((line = bufferedReader.readLine()) != null) {
                lines.add(line);
            }
            bufferedReader.close();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        return lines;
    }
}

public class MyCustomThread implements Runnable
{

    String line;

    MyCustomThread(String line)
    {
        this.line = line;
    }

    @Override
    public void run()
    {
        System.out.println(Thread.currentThread().getName() + " processed line:" + line);

    }

}

编辑：此实现不会阻止ExecutorService提交。我的意思是，无论之前提交的MyCustomThreads是否已完成，都会为文件中的每一行创建一个新的MyCustomThread实例。您可以添加一个阻塞/限制工作队列来防止出现这种情况。

ExecutorService executor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LimitedQueue<Runnable>(10));

可以在此处找到阻塞/限制队列实现的示例：

Answer 2

你做错了一点。执行器的原理是将工作单元实现为可运行或可调用（而不是线程）。每个Runnable或Callable都应该做一个原子工作，这与其他Runnable或Callable是互斥的。

Executor服务在内部使用线程池，因此创建线程组和线程没有任何好处。

试试这个简单的片段：

ExecutorService executor = Executors.newFixedThreadPool(5);`
executor.execute(new MyRunnableWorker());

public class MyRunnableWorker implements Runnable{
    private String ip;
    private String threadName;
    private int threadTimeout = 10;

    public MyRunnableWorker(String ip){
        this.ip = ip;
    }

    public MyRunnableWorker(String ip,int threadTimeout,String threadName){
        this.ip = ip;
        this.threadTimeout = threadTimeout;
        this.threadName = threadName;

        System.out.prinln("MyRunnableWorker constructor has been called!");
    }

    @Override
    public void run(){    {
        // do some stuff that takes time ....
    }
}

这将为您提供您想要的东西。还要尝试使用Visual alVM测试线程代码执行情况，以查看线程如何运行以及负载分布情况。