Spring Batch : Tasklet mit multi-threaded Testamentsvollstrecker hat sehr schlechte Leistungen im Zusammenhang Drosselung Algorithmus

Als Alex sagte, es scheint dieses Verhalten ein Vertrag nach javadocs von :

Unterklassen brauchen, um eine Methode, die bekommt die nächste Folge * und eins, das wartet, bis alle Ergebnisse werden zurückgegeben, auch aus gleichzeitig * Prozesse oder threads

Betrachten:

TaskExecutorRepeatTemplate#waitForResults

Weitere option für Sie wäre, um die Partitionierung verwenden :

• Eine TaskExecutorPartitionHandler, die ausgeführt wird, Elemente aus Partitionned ItemReader, siehe unten
• Einem Partitioner-Implementierung, gibt die Bereiche an, die verarbeitet werden ItemReader, siehe ColumnRangePartitioner unten
• Eine CustomReader das Lesen von Daten mithilfe von was-Partitionierer wird gefüllt haben, sehen myItemReader Konfiguration unter

Michael Minella erklärt in Kapitel 11 seines Buches Pro Spring Batch:

<batch:job id="batchWithPartition">
    <batch:step id="step1.master">
        <batch:partition  partitioner="myPartitioner" handler="partitionHandler"/>
    </batch:step>       
</batch:job>
<!-- This one will create Paritions of Number of lines/ Grid Size--> 
<bean id="myPartitioner" class="....ColumnRangePartitioner"/>
<!-- This one will handle every partition in a Thread -->
<bean id="partitionHandler" class="org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler">
    <property name="taskExecutor" ref="multiThreadedTaskExecutor"/>
    <property name="step" ref="step1" />
    <property name="gridSize" value="10" />
</bean>
<batch:step id="step1">
        <batch:tasklet transaction-manager="transactionManager">
            <batch:chunk reader="myItemReader"
                writer="manipulatableWriterForTests" commit-interval="1"
                skip-limit="30000">
                <batch:skippable-exception-classes>
                    <batch:include class="java.lang.Exception" />
                </batch:skippable-exception-classes>
            </batch:chunk>
        </batch:tasklet>
</batch:step>
 <!-- scope step is critical here-->
<bean id="myItemReader"    
                        class="org.springframework.batch.item.database.JdbcCursorItemReader" scope="step">
    <property name="dataSource" ref="dataSource"/>
    <property name="sql">
        <value>
            <![CDATA[
                select * from customers where id >= ? and id <=  ?
            ]]>
        </value>
    </property>
    <property name="preparedStatementSetter">
        <bean class="org.springframework.batch.core.resource.ListPreparedStatementSetter">
            <property name="parameters">
                <list>
 <!-- minValue and maxValue are filled in by Partitioner for each Partition in an ExecutionContext-->
                    <value>{stepExecutionContext[minValue]}</value>
                    <value>#{stepExecutionContext[maxValue]}</value>
                </list>
            </property>
        </bean>
    </property>
    <property name="rowMapper" ref="customerRowMapper"/>
</bean>

Partitioner.java:

 package ...;
  import java.util.HashMap;  
 import java.util.Map;
 import org.springframework.batch.core.partition.support.Partitioner;
 import org.springframework.batch.item.ExecutionContext;
 public class ColumnRangePartitioner  implements Partitioner {
 private String column;
 private String table;
 public Map<String, ExecutionContext> partition(int gridSize) {
    int min =  queryForInt("SELECT MIN(" + column + ") from " + table);
    int max = queryForInt("SELECT MAX(" + column + ") from " + table);
    int targetSize = (max - min) / gridSize;
    System.out.println("Our partition size will be " + targetSize);
    System.out.println("We will have " + gridSize + " partitions");
    Map<String, ExecutionContext> result = new HashMap<String, ExecutionContext>();
    int number = 0;
    int start = min;
    int end = start + targetSize - 1;
    while (start <= max) {
        ExecutionContext value = new ExecutionContext();
        result.put("partition" + number, value);
        if (end >= max) {
            end = max;
        }
        value.putInt("minValue", start);
        value.putInt("maxValue", end);
        System.out.println("minValue = " + start);
        System.out.println("maxValue = " + end);
        start += targetSize;
        end += targetSize;
        number++;
    }
    System.out.println("We are returning " + result.size() + " partitions");
    return result;
}
public void setColumn(String column) {
    this.column = column;
}
public void setTable(String table) {
    this.table = table;
}
}

In meinem Fall, wenn ich nicht die throttle-Grenze, dann nur 4 threads kommen in der read () - Methode der ItemReader, was auch der Standardwert für die Anzahl der threads, wenn nicht anders angegeben in tasklet-tag als pro Spring Batch Dokumentation.

Wenn ich hier mehr threads.e.g 10 oder 20 oder 100, dann nur 8 threads kommen in der read () - Methode der ItemReader