ThreadPoolExecutor正在运行的应用程序

Question

账单类具有与国家明智账单相关的所有逻辑。它从数据库中获取结果，然后为用户开票。帐单类实现Runnable。我想根据国家参数并行执行结算，以便大量用户（500万+）非常快速地结算。现在需要数小时完成。

我想实现ThreadPoolExecutor执行Billing类，但很困惑如何？以下有什么区别或者我做错了什么？请建议！区共有20个国家，但我在这里粘贴仅5

//for 20 countries ThreadPoolExecutor (20,20,20.......)???? 

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 5, TimeUnit.SECONDS, 
new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.CallerRunsPolicy()); 

executor.execute(new Billing("UK")); 
executor.execute(new Billing("USA")); 
executor.execute(new Billing("Germany")); 
executor.execute(new Billing("Spain")); 
executor.execute(new Billing("Italy")); 

OR

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 5, TimeUnit.SECONDS, 
new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.CallerRunsPolicy()); 
for(int i=0;i<5;i++) // for 20 countries i<20?? 
{  

    executor.execute(new Billing("UK")); 
    executor.execute(new Billing("USA")); 
    executor.execute(new Billing("Germany")); 
    executor.execute(new Billing("Spain")); 
    executor.execute(new Billing("Italy")); 
} 

while (! executor.isTerminated()) { 
    try{ 
     executor.awaitTermination(100, TimeUnit.SECONDS); 
    }catch(InterruptedException iE) 
    { 
     iE.printStackTrace(); 
     System.out.println("Executor Exception: "+ iE); 
    } 

在此先感谢！

Answer 1

循环解决方案看起来不正确。不需要多次执行相同的Runnable。

你既corePoolSize和maximumPoolSize设置为5，这意味着执行程序将保持线程的数量在游泳池5，即使它们是空闲实例ThreadPoolExecutor。它还说池中的线程数不能超过5。

在此位置，您可以预计至多5线程正在并行执行任务（Billing对象）。

随着您继续提交Billing对象到executor与execute方法它们被添加到您提供的ArrayBlockingQueue。此队列的大小为10。在某些情况下，队列中的队列可能已经处于最大容量并且不能承担更多任务，在这种情况下，任务将被拒绝并提供给ThreadPoolExecutor构造函数中提供的RejectedExecutionHandler。它的工作是用已实施的方法rejectedExecution处理被拒绝的任务。

如果你想找到是否有任何被拒绝的任务，你必须提供自己的RejectedExecutionHandler而不是使用默认的ThreadPoolExecutor.CallerRunsPolicy。你可以像这样做：

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 5, 
     TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), 
     new RejectedExecutionHandler() { 
      @Override 
      public void rejectedExecution(Runnable r, 
        ThreadPoolExecutor executor) { 
       System.out.println("I got rejected: " + r); 
       if (!executor.isShutdown()) { 
        r.run(); 
       } 
      } 
     }); 

Answer 2

我不确定你了解循环如何工作。不同之处在于第二个代码块将在每个列出的国家/地区运行计费5次。

Answer 3

假设你正在谈论的代码for环部分并不明显，将如何工作。

理想的循环将是这个样子：

for(String country : countryCollection) { 
    executor.execute(new Billing(country)); 
} 

Answer 4

你有没有考虑使用enum？

static class Billing implements Runnable { 

    enum Country { 
    UK, 
    USA, 
    Germany, 
    Spain, 
    Italy; 
    } 

    public Billing(Country country) { 
    } 

    @Override 
    public void run() { 
    } 
} 
public void test() { 
    ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 5, TimeUnit.SECONDS, 
     new ArrayBlockingQueue<Runnable>(10), 
      new ThreadPoolExecutor.CallerRunsPolicy()); 
    for (Billing.Country country : Billing.Country.values()) { 
    executor.execute(new Billing(country)); 
    } 
} 

Answer 5

第一：忘记循环

for(int i=0;i<5;i++) // for 20 countries i<20?? 
{  

    executor.execute(new Billing("UK")); 
    executor.execute(new Billing("USA")); 
    executor.execute(new Billing("Germany")); 
    executor.execute(new Billing("Spain")); 
    executor.execute(new Billing("Italy")); 
} 

这一切账单多次遍历。

做的正确的事情是在第一个片段：

executor.execute(new Billing("UK")); 
executor.execute(new Billing("USA")); 
executor.execute(new Billing("Germany")); 
executor.execute(new Billing("Spain")); 
executor.execute(new Billing("Italy")); 

另一个错误在于对终端的检查：

while (! executor.isTerminated()) { 
    try{ 
     executor.awaitTermination(100, TimeUnit.SECONDS); 
    }catch(InterruptedException iE) 
    { 
     iE.printStackTrace(); 
     System.out.println("Executor Exception: "+ iE); 
    } 
} 

的Executor.awaitTermination的Javadoc说：

块直到所有任务在关闭请求后完成执行，

但您永远不会发出关闭请求。

在你的情况，你可以充分利用ExecutorCompletionService为：

CompletionService<String> ecs = new ExecutorCompletionService<String>(executor); 
List<String> countries= Arrays.asList("UK","USA","Germany","Spain","Italy"); 
for(String country : countries) { 
    ecs.submit(new Billing(country),country); 
} 
// wait for completion 
for(int i=0;i<countries.size();i++){ 
     ecs.take(); // wait for next country completion 
} 
// all work completed, shutdown 
executor.shutdownNow(); 

Answer 6

另一种方式来思考实现这个就是看看The Fork/Join Framework。这看起来似乎可以真正从炼钢获益。例如，你可以看起来相当干净地分解它。这基本上可以让您打破用户或部分用户的计费任务，而不是让一个看起来代表一个国家的线程在整个计费过程中工作。

你可以找到的链接库：here if you are using a version of Java < 7