ThreadPoolExecutor Запуск приложения

Question

Класс фактурирования имеет всю логику, связанную с биллированием по стране. Он принимает результат из базы данных, а затем оплачивает пользователей. Класс Billing реализует Runnable. Я хочу выполнить биллинг параллельно в соответствии с параметром страны, так что огромное количество пользователей (5 миллионов +) выставляется очень быстро. Теперь для завершения требуется несколько часов и часов.

Я пытаюсь реализовать 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")); 

ИЛИ

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 несколько раз.

Вы инстанцирование ThreadPoolExecutor как с corePoolSize и maximumPoolSize набором для 5, что означает, что исполнитель будет поддерживать количество потоков в бассейне 5, даже если они находятся в режиме ожидания. В нем также говорится, что в пуле может быть не более 5 потоков.

При этом на месте можно ожидать не более 5 потоков, выполняющихся параллельно выполнению задач (Billing объектов).

Как вы продолжить представление Billing объектов executor с execute методами, который они добавляют к ArrayBlockingQueue, предоставленным Вами. Размер этой очереди: 10. Возможно, что в какой-то очередь очередь уже имеет максимальную емкость и не может выполнять больше задач, и в этом случае задача отклоняется и передается RejectedExecutionHandler, предоставленному в конструкторе ThreadPoolExecutor. Его задача - обработать отклоненные задачи с помощью реализованного метода 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); 
    } 
} 

Javadoc из Executor.awaitTermination говорит:

блоков пока все задачи не завершили выполнение после запроса на завершение работы,

, но вы никогда не выдаете запрос на завершение работы.

В вашем случае, вы могли бы использовать самые 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