java - 如何在java中为每个线程分配一定数量的rdms记录进行处理

Question

在这里，我创建了两个线程。我尝试批量从 mysql 读取记录，并尝试将该批处理分配给每个线程，以便将这些记录写入控制台。

假设我有两个批处理，每个批处理包含 10 条记录。我尝试将每个批处理分配给每个线程，以便将这些记录写入控制台。

这里只有thread1正在读取所有1000条记录(每批10条记录)。但是我要创建thread2。如何创建thread2，使其参与并行处理。

我的代码:

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 !!!");
    }

}

我的可运行:

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());
        }


    }

}

客户数据:

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

我经常遵循的一个快速解决方案，

有内部并发队列。将批处理放入其中。 Foreach 线程使值出队(从内部队列中批量)并分配它。

不是完整的解决方案

第 1 步:

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

第2步:为每个选定的项目创建一批对象(比如10个)

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

第三步:

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

第四步:

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