Wie eine bestimmte Anzahl von RDM-Datensätzen jedem Thread zur Verarbeitung in Java zugewiesen wird

Question

Hier habe ich zwei Threads erstellt. Ich versuche, die Datensätze von MySQL in einem Stapel zu lesen und versuchen, diesen Stapel jedem Thread zum Schreiben zuzuweisen diese Aufzeichnungen zu trösten.

Angenommen, ich habe zwei Stapel mit je 10 Datensätzen. Ich versuche, jeden Stapel jedem Thread zuzuordnen, um diese Datensätze in die Konsole zu schreiben.

Hier liest thread1 nur alle 1000 Datensätze (10 Datensätze pro Batch). Aber ich soll thread2 erstellen. Wie kann ich thread2 erstellen, so dass es parallel dazu geht.

Mein Code:

public class DataRead { 

    static final String JDBC_DRIVER = "com.mysql.jdbc.Driver"; 
    static final String DB_URL = "jdbc:mysql://localhost/test"; 

    static final String USER ="root"; 
    static final String PASS ="abc"; 

    public static void main(String[] args) throws ClassNotFoundException, SQLException { 
     Connection conn = null; 
     Statement stmt = null; 

     Class.forName(JDBC_DRIVER); 

     System.out.println("Connecting to a selected database"); 
     conn = DriverManager.getConnection(DB_URL, USER,PASS); 
     System.out.println("Connected to Database Successfully !!!"); 

     ResultSet rs = null; 

     stmt = conn.createStatement(); 
     int maxRecords = 10; 
     int initialcount =0; 
     int count =0; 
     rs = stmt.executeQuery("SELECT COUNT(*) FROM customer"); 
     rs.next(); 
     count = rs.getInt(1); 
     System.out.println(count); 

     for(int i=0;i<count/10;i++){ 

       String sql =" SELECT customerId,firstName,lastName FROM customer ORDER BY customerId LIMIT "+initialcount+","+maxRecords; 

       rs = stmt.executeQuery(sql); 

       String CustID= null; 
       String firstName = null; 
       String lastName = null; 


       ExecutorService threadExecutor = Executors.newFixedThreadPool(2); 
       List<CustomerData> list = new ArrayList<CustomerData>(); 

       while(rs.next()){ 

        CustomerData customer = new CustomerData(); 
        customer.setCustID(rs.getString("customerId")); 
        customer.setFirstName(rs.getString("firstName")); 
        customer.setLastName(rs.getString("lastName")); 

        list.add(customer); 

       } 
       MyRunnable task1 = new MyRunnable(list); 
        threadExecutor.execute(task1); 


       threadExecutor.shutdown(); 
       while(!threadExecutor.isTerminated()){ 
        try { 
         threadExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); 
         } catch (InterruptedException e) { 

         } 
       } 

       initialcount = initialcount + maxRecords; 
     } 

    System.out.println("Finished all threads !!!"); 
    } 

} 

MyRunnable:

public class MyRunnable implements Runnable{ 

    private final List list; 

    public MyRunnable(List list){ 
     this.list = list; 
    } 

    @Override 
    public void run() { 
     for(int i=0;i<list.size();i++){ 
      CustomerData cust = (CustomerData) list.get(i); 
      System.out.println(Thread.currentThread().getName()+" "+"Hello: "+cust.getCustID()+" "+cust.getFirstName()+" "+cust.getLastName()); 
     } 


    } 

} 

Customer:

public class CustomerData { 

     private String custID; 
     private String firstName; 
     private String lastName; 

    public String getCustID() { 
     return custID; 
    } 
    public void setCustID(String custID) { 
     this.custID = custID; 
    } 
    public String getFirstName() { 
     return firstName; 
    } 
    public void setFirstName(String firstName) { 
     this.firstName = firstName; 
    } 
    public String getLastName() { 
     return lastName; 
    } 
    public void setLastName(String lastName) { 
     this.lastName = lastName; 
    } 
} 

Answer 1

Eine schnelle Lösung, die ich oft folgen verwendet,

internehaben. Fügen Sie Charge hinein. Foreach thread löscht den Wert (Batch aus der internen Warteschlange) und weist ihn zu.

keine vollständige Lösung

Schritt 1:

Queue<List<CustomerData>> queue = new ConcurrentLinkedQueue<ArrayList<CustomerData>>(); 

Schritt 2: foreach ausgewähltes Element erstellen Charge von Objekten (etwa 10)

ArrayList<CustomerData> listofobjects = // insert batch of vales selected from DB 

step3:

queue.add(listofobjects); // add your batches into concurrent queue 

Schritt 4:

while(queue.isEmpty()){ 
    ArrayList<CustomerData> batch = queue.poll(); 
    // Pass batch to Thread and print it. 
}