java - 如何改进运行异步任务

Question

我有每 1 分钟运行一次的任务。该任务包含几个处理一些数据的方法。进程 1 读取状态 1 中的数据(一个或多个)，并在该过程结束时将状态更新为 2。进程 2 读取状态 2 中的数据，依此类推。所以现在我想使用 taskExecutor (org.springframework.core.task.SimpleAsyncTaskExecutor) 来并行运行它来改进此处理:

public void process1() {

    List<Object> objects = someDao.readDataWithStatus("1");
    if (objects == null || objects.isEmpty()) {
        return;
    }

    for (final Object object : objects) {
            if (BooleanUtils.isTrue(isParalelProcess())) {
                taskExecutor.execute(new Runnable() {

                    @Override
                    public void run() {
                        process(object);
                    }
                });
            } else {
                process(object);
            }           
    }
}

假设我们有一个状态为 1 的对象，并且我们想要并行处理。每一分钟运行的任务开始处理。 Process1读取状态为1的数据放入taskExecutor中，进入下一个方法。此方法未找到状态为 2 的对象，因此此处无需执行任何操作。一分钟后，process2 读取状态为 2 的对象，依此类推。处理对象需要 1 秒到几分钟。正如您所看到的，过程明显减慢，只需要几秒钟。是否可以选择如何改进此处理？

Answer 1

我花了很少的时间在上面。也许对你有帮助。

public class SampleTest
{

    public static void main(String[] args)
    {

        int noOfThreads = 100;
        int maxThreadinQue = 100;

        LinkedBlockingQueue<Runnable> processOneworkQueue = new LinkedBlockingQueue<Runnable>();

        ThreadPoolExecutor processOneThreadPoolExecutor = new ThreadPoolExecutor(noOfThreads, noOfThreads, 0L,
                TimeUnit.SECONDS,

                processOneworkQueue, new ThreadFactory()
                {
                    private AtomicInteger itsCounter = new AtomicInteger();

                    public Thread newThread(Runnable theRunnable)
                    {
                        Thread aThread = new Thread(theRunnable, "theThreadName" + "#" + itsCounter.getAndIncrement());
                        aThread.setDaemon(true);
                        return aThread;
                    }
                });

        LinkedBlockingQueue<Runnable> processTwoworkQueue = new LinkedBlockingQueue<Runnable>();

        ThreadPoolExecutor processTwoThreadPoolExecutor = new ThreadPoolExecutor(noOfThreads, noOfThreads, 0L,
                TimeUnit.SECONDS,

                processTwoworkQueue, new ThreadFactory()
                {
                    private AtomicInteger itsCounter = new AtomicInteger();

                    public Thread newThread(Runnable theRunnable)
                    {
                        Thread aThread = new Thread(theRunnable, "theThreadName" + "#" + itsCounter.getAndIncrement());
                        aThread.setDaemon(true);
                        return aThread;
                    }
                });

        ArrayList<SampleObject> sampleObjects = new ArrayList<SampleObject>();
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());
        sampleObjects.add(new SampleObject());

        while (true)
        {
            Iterator<SampleObject> it = sampleObjects.iterator();
            while (it.hasNext())
            {

                final SampleObject sampleObject = it.next();

                if (sampleObject.getStatus() == 0)
                {
                    if (processOneworkQueue.size() < maxThreadinQue)
                    {
                        processOneThreadPoolExecutor.submit(new Runnable()
                        {
                            @Override
                            public void run()
                            {
                                process1(sampleObject);

                            }
                        });
                    }
                }
                else if (sampleObject.getStatus() == 1)
                {
                    if (processTwoworkQueue.size() < maxThreadinQue)
                    {
                        processTwoThreadPoolExecutor.submit(new Runnable()
                        {
                            @Override
                            public void run()
                            {
                                process2(sampleObject);

                            }
                        });
                    }
                }
                else if (sampleObject.getStatus() == 2)
                {

                    it.remove();

                }
            }
        }
    }

    static void process1(SampleObject sampleObject)
    {
        System.out.println("process1");
        sampleObject.setStatus(1);

    }

    static void process2(SampleObject sampleObject)
    {
        System.out.println("process2");
        sampleObject.setStatus(2);
    }

}

class SampleObject
{
    int status=0;

    public int getStatus()
    {
        return status;
    }

    public void setStatus(int status)
    {
        this.status = status;
    }
}