【java/Thread】用无尽线程和固定大小线程池循环分批处理数据例程

【需求】

设计一循环线程，分批并行处理emp5202中的数据，要求并行数可以配置。

这是为一个实际工程制作的概念设计，两者流程一致，只是处理数据量和粒度不同。

【实现】

采用无尽循环LoopThread实现循环。

采用Executors.newFixedThreadPool（N）来进行并行处理。

【流程图】

看懂了这幅图就明白了代码意图。

【具体代码】 

LoopThread类：

package com.hy.lab.loopth;

import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class LoopThread extends Thread{

    public void run(){
        DBAccessor accessor=new DBAccessor();
        ExecutorService es= Executors.newFixedThreadPool(2);
        Random rnd=new Random();
        while(true){
            int count=accessor.getCount();

            if(count>0) {
                List emps=accessor.getEmpList();
                accessor.deleteAllEmp();
                for(Emp emp:emps){
                    es.submit(new RunThread(emp));
                }
            }
        }
    }

    public static void main(String[] args){
        new LoopThread().start();
    }
}

RunThread类：

package com.hy.lab.loopth;

public class RunThread implements Runnable{
    private Emp emp;

    public RunThread(Emp emp){
        this.emp=emp;
    }

    @Override
    public void run() {
        System.out.println(this.emp.name+" handled.");
        try {
            Thread.sleep(1*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

DBAccesor类：

package com.hy.lab.loopth;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;

class Emp{
    public Emp(int id,String name){
        this.id=id;
        this.name=name;
    }

    int id;
    String name;
}

public class DBAccessor {
    public int getCount(){
        String sql="select count(*) from emp5202";
        Connection conn =DsUtil.getConn();

        try(PreparedStatement pstmt =conn.prepareStatement(sql);){

            ResultSet rs=pstmt.executeQuery();
            while(rs.next()){
                return rs.getInt(1);
            }
        }catch(Exception e){
            e.printStackTrace();
        }

        return 0;
    }

    public List getEmpList(){
        String sql="select id,name from emp5202 order by hdate desc";
        Connection conn =DsUtil.getConn();

        try(PreparedStatement pstmt =conn.prepareStatement(sql);){
            List emps=new ArrayList<>();

            ResultSet rs=pstmt.executeQuery();
            while(rs.next()){
                int id=rs.getInt(1);
                String name=rs.getString(2);

                emps.add( new Emp(id,name));
            }

            return emps;
        }catch(Exception e){
            e.printStackTrace();
        }

        return null;
    }

    public boolean deleteAllEmp(){
        String sql="delete from emp5202 where 1=1";
        Connection conn =DsUtil.getConn();

        try(PreparedStatement pstmt =conn.prepareStatement(sql);){

            pstmt.execute();
            return true;
        }catch(Exception e){
            e.printStackTrace();
        }

        return false;
    }
}

DsUtil类：

package com.hy.lab.loopth;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

public class DsUtil {
    private static Connection conn;

    static{
        conn=createConnection();
    }

    public static Connection getConn(){
        return conn;
    }

    private static Connection createConnection() {
        Connection conn = null;
        try {
            Class.forName("oracle.jdbc.driver.OracleDriver");
            String url = "jdbc:oracle:thin:@127.0.0.1:1521:orcl";
            String user = "luna";
            String pass = "1234";
            conn = DriverManager.getConnection(url, user, pass);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return conn;
    }
}

输出：

OUJYUSIV handled.
VJJOFHFCBKHOX handled.
KRXVJHPAGBL handled.
KCJIPJFXBNIBIQAH handled.
DJQLHRYVHT handled.
GBPYMWYQTVNFBOB handled.
VMCGWPRSAQESPD handled.
SIHQZPQJL handled.
HCGCPRN handled.
IVZTZBC handled.
GTHUBVXPMBRGUYIGMB handled.
YSLPEXRVWBV handled.
ELBMUBTRFHLEMXUXVN handled.
ANIQORJCNE handled.
IHTDXHUFC handled.
NJZMTORACI handled.
XBDXSTXPAHHQRX handled.
SKPXIUMPSGKS handled.
QBJKIKTCFPGQSQG handled.
NFDAZFCOXNJNWYMTKXS handled.
MAOAHIUDYLUANEZ handled.
BQFFGERQY handled.
CSNXJVXW handled.
MMAEKFETEMJR handled.
EVFZVHKIR handled.
NCTTQJFRBMSPQAVBAXM handled.
HVTKEBIZWODDW handled.
QYSDRBPWIWFDYTNOLQ handled.
IWXQYYOGGIVKKXEQTQT handled.

建表及填充语句：

create table emp5202(
    id number(10),
    name nvarchar2(20),
    hdate timestamp
);

insert into emp5202
select rownum,dbms_random.String('*',dbms_random.value(6,20)),to_date('20000101','yyyyMMdd')+rownum
from dual
connect by level<11;

END