Mehrere schedule target sendestasks RejectedExecutionException

Question

In meiner Anwendung habe ich mehrere scheduler Threads, die Aufgaben erstellen. Zum Beispiel jeweils Scheduler-Threads können Bündel von Aufgaben erstellen:

TaskCreator tastCreator; 
for (Report report: report) { 
    taskCreator.createTask(report); 
} 

Die Scheduler-Threads gleichzeitig ausgeführt werden können, wie Sie aus den Protokollen sehen:

15:57:20.107 INFO [ scheduler-4] c.task.ReportExportSchedulerTask : Task created 
15:57:20.107 INFO [ scheduler-2] c.task.ReportExportSchedulerTask : Task created 

Ich habe eine TaskCreator Komponente wie folgt, dass Pässe die Aufgabe der executeJob():

@Component 
public class TaskCreator { 
    @Autowired 
    private SftpTaskExecutor sftpTaskExecutor; 
    @Autowired 
    SftpConfig sftpConfig; 

    @Autowired 
    private SFTPConnectionManager connectionManager; 

    public void createTask(Report report) { 
     sftpTaskExecutor.executeJob(new JobProcessorTask(...)); 
    } 

    public void validateTasksExecution() { 
     sftpTaskExecutor.getExecutorService().shutdown(); 
     while (!sftpTaskExecutor.getExecutorService().isTerminated()) ; 
     connectionManager.disconnect(); 
    } 
} 

SftpTaskExecutorComponent wie folgt, dass konstruiert ein executorService auf die ich die oben genannten Aufgaben zu übermitteln:

@Component 
public class SftpTaskExecutor { 
    private ExecutorService executorService = Executors.newSingleThreadExecutor(); 

    public void executeJob(JobProcessorTask jobProcessorTask) { 
     executorService.execute(jobProcessorTask); 
    } 

    public ExecutorService getExecutorService() { 
     return executorService; 
    } 
} 

meine Frage, wenn zwei oder mehr scheduler Threads Aufgaben erstellen und Einreichen Dienst gleichzeitig Executer, wirft der oben eine RejectedExecutionException wenn eine Scheduler-Task nicht beendet ist (d. h Datei nicht gesendet)

Für jede Zeitplan Threads, ich muss validateTasksExecution() aufrufen können, ohne den anderen Scheduler-Thread zu stören. Mit anderen Worten, nicht trennen, während der andere Scheduler noch verarbeitet.

Bin ich die ExecutorService in dieser Hinsicht richtig verwenden? Wie kann ich das oben genannte ändern, um threadsicher zu sein?

Answer 1

Meine Frage ist, wenn zwei oder mehrere Scheduler-Threads Aufgaben Erstellen und Einreichen Dienst gleichzeitig Testamentsvollstrecker wirft die oben eine RejectedExecutionException mit einem Scheduler-Task nicht beendet (dh nicht gesendeten Datei)

Lassen Sie uns werfen Sie einen Blick auf die javadocs for ExecutorService.execute(...)

RejectedExecutionException - wenn diese Aufgabe nicht zur Ausführung akzeptiert werden kann.

am ThreadPoolExecutor (und zugehörigen Code) Bei der Suche, erhalten die Arbeitsplätze aus 2 Gründen abgelehnt:

• Die Warteschlange für die Aufträge voll ist (dies für Sie nicht gilt, weil die Warteschlangen standardmäßig unbegrenzt)
• der Testamentsvollstrecker Dienst wird nicht mehr (ding ding ding) läuft

ich glaube, dass Ihr Testamentsvollstrecker Service heruntergefahren wurde, wahrscheinlich, weil die erste Ihrer Fäden hat genannt validateTasksExecution() vor dem 2. Thread ruft executeJob(...). Ihr Code ist falsch, wenn Sie versuchen, diesen Thread-Pool wiederzuverwenden. Dass Sie auch die connectionManager() schließen, frage mich, ob Sie die SftpTaskExecutor überhaupt wiederverwenden wollen.

Wenn Sie wollen, dass jeder Thread zu sehen, ob ihr Betrieb durchgeführt wird, haben aber den Thread-Pool Aufenthalt läuft, dann müssen, werden Sie das Future(s) vom ExecutorService.submit(...) Methode Speichern und get() auf sie nennen. Das wird Ihnen sagen, wenn die Jobs erledigt sind.

Etwas wie:

public Future<Void> createTask(Report report) { 
    return sftpTaskExecutor.executeJob(new JobProcessorTask(...)); 
} 

public void validateTasksExecution(Future<Void> future) { 
    // there is some exceptions here you need to handle 
    future.get(); 
} 

public void shutdown() { 
    sftpTaskExecutor.shutdown(); 
    connectionManager.disconnect(); 
} 

... 

public Future<Void> executeJob(JobProcessorTask jobProcessorTask) { 
    return executorService.submit(jobProcessorTask); 
} 

Wenn Sie mehrere Aufträge überwachen müssen, dann sollten Sie sie in einer Sammlung speichern und get() auf sie seriell nennen, obwohl die Aufträge parallel ausgeführt werden soll.

Die Alternative wäre für Sie eine separate ExecutorService für jede Transaktion, die verschwenderisch ist, aber vielleicht nicht so schlecht, wenn man bedenkt, dass es Sftp-Anrufe verwaltet.

while (!sftpTaskExecutor.getExecutorService().isTerminated()) ; 

Ja, Sie wollen nicht, wie das spinnen. Siehe awaitTermination(...) javadocs.