BigData

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.sql.DataSource;

import org.apache.commons.lang.StringUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class InActiveCustomer {
?private static? String RAPID_CUSTOMER_QUERY = "SELECT TOP 3 A.GFCID FROM "
???+ "VA_CUS_CUSTOMER A left join VA_CUS_RISK_ATTRIBUTE B on A.CUSTOMER_ID=B.CUSTOMER_ID "
???+ "WHERE B.CUSTOMER_ID IS NULL AND A.GFCID LIKE '000001%' AND A.CAGID=A.GFCID MAXGFCID ORDER BY GFCID ASC";

?private final String RISKODS_CUSTOMER_QUERY = "SELECT C.CUSTOMER_ID,CUSTOMER_NAME,CAGID,GFPID,GFCID "
???+ " FROM V_CUS_CUSTOMER C,V_CUS_CONFIDENTIAL_RATING,V_CUS_RISK_ATTRIBUTE "
???+ " WHERE C.CUSTOMER_ID = V_CUS_CONFIDENTIAL_RATING.CUSTOMER_ID AND "
???+ " C.CUSTOMER_ID = V_CUS_RISK_ATTRIBUTE.CUSTOMER_ID AND C.GFCID in? ";

?public DataSource rapidDataSource;
?public DataSource odsDataSource;

?private void findInRapid(ExecutorService service) throws SQLException {
??Connection conn1 = rapidDataSource.getConnection();
??int times = 1;
??String sqlGfcid = "";
??while(true){
???PreparedStatement ps1 = conn1.prepareStatement(rapidSql(times,sqlGfcid));
???ResultSet rs1 = ps1.executeQuery();
???//rs1.setFetchSize(1000);
???List<String> elements = new ArrayList<String>();
???while (rs1.next()) {
????elements.add(rs1.getString(1));
????if (elements.size() == 3) {
?????StringBuilder sql = new StringBuilder("");
?????for (int i = 0; i < elements.size(); i++) {
??????Long gfcid = Long.valueOf(elements.get(i));
??????if (i != elements.size() - 1) {
???????sql.append("'" + gfcid + "'" + ",");
??????} else {
???????sql.append("'" + gfcid + "'");
???????sqlGfcid = StringUtils.leftPad(gfcid.toString(),15,'0');
??????}
?????}
//?????find_RISKODS(service,
//???????RISKODS_CUSTOMER_QUERY + "(" + sql.toString() + ")");
?????elements.clear();
????}
???}
???times++;
??}
?}

?private void find_RISKODS(ExecutorService service,String sql)
???throws SQLException {
??CustomerThread ct = new CustomerThread(odsDataSource,sql);
??service.execute(ct);

?}
?
?private static String rapidSql(int times,String lastGfcid){
??String rapidSql;
??if(times == 1){
??? rapidSql = SqlBuilder(RAPID_CUSTOMER_QUERY,"MAXGFCID","");
??}else{
??? rapidSql = SqlBuilder(RAPID_CUSTOMER_QUERY," AND GFCID > '"+lastGfcid+"' ");
??}
??return rapidSql;
?}
?
?private static String SqlBuilder(String sql,String replace,String newString){
??String temp = sql;
??String result = temp.replace(replace,newString);
??return result;
?}

?public static void main(String[] args) throws SQLException {
??ExecutorService es = Executors.newFixedThreadPool(3);
??InActiveCustomer inActiveCustomer = new InActiveCustomer();
??ApplicationContext context = new ClassPathXmlApplicationContext("db-context2.xml");
??inActiveCustomer.rapidDataSource = (DataSource) context.getBean("rapidDataSource");
??inActiveCustomer.odsDataSource = (DataSource) context.getBean("odsDataSource");
??inActiveCustomer.findInRapid(es);

?}
}

?

---------------------------------------------------------------

?

import java.io.FileWriter;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import javax.sql.DataSource;

public class CustomerThread implements Runnable {

?public CustomerThread(DataSource odsDataSource,String sql) {
??super();
??this.odsDataSource = odsDataSource;
??this.sql = sql;
?}

?public DataSource odsDataSource;
?public String sql;
?Connection con = null;
?PreparedStatement ps = null;
?ResultSet rs = null;
?List<String> result = new ArrayList<String>();

?@Override
?public void run() {
??try {
???// System.out.println(Thread.currentThread().getName()+"_"+new
???// Date());
???con = odsDataSource.getConnection();
???// System.out.println(Thread.currentThread().getName()+"_"+con);
??? System.out.println(Thread.currentThread().getName()+"_"+sql.substring(230));
???ps = con.prepareStatement(sql);
???// ps.setFetchSize(1000);
???rs = ps.executeQuery();
???//rs.setFetchSize(1000);
???while (rs.next()) {
????String element = rs.getString(1);
????result.add(" "+element);
????System.out.println(" The same element " + element);
???}
//???FileWriter fileWriter = new FileWriter("c:PersonResult.txt");
//???for (int i = 0; i < result.size(); i++) {
//????fileWriter.write(String.valueOf(result.get(i)));
//???}
//???fileWriter.flush();
//???fileWriter.close();
???// System.out.println(Thread.currentThread().getName()+"_"+new Date());
??} catch (SQLException e) {
???e.printStackTrace();
//??} catch (IOException e) {
//???// TODO Auto-generated catch block
//???e.printStackTrace();
??} finally {
???try {
????rs.close();
????ps.close();
????con.close();
???} catch (SQLException e) {
????e.printStackTrace();
???}

??}
?}

}

?http://www.flickr.com/photos/jiutong