schedule.java.svn-base

数据库通用适配器：实现行粒度的数据库的转换

package hdaty.prtds.rdgadaptor.gui;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;

	enum Cycle{
		YEAR,
		MONTH,
		WEEK,
		DAY,
		HOUR,
		MINUTE;
	}
public class Schedule {

	public static ScheduledThreadPoolExecutor scheduler;
	
	public  Schedule(){
		scheduler = new ScheduledThreadPoolExecutor(10);
		
	}

	public Schedule(int size){
		scheduler = new ScheduledThreadPoolExecutor(size);
	}
	
	/**
	 *周期性执行任务
	 * @param event 需要执行的任务，该任务必须实现runnable接口
	 * @param date 任务开始执行的时间,格式为“yyyy/MM/DD hh:mm:ss”
	 */
	public void repeat(Runnable event,String date)
	throws InterruptedException, ParseException{
		long delay = 0;
		SimpleDateFormat fTime = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
	    Date starttime = fTime.parse(date);
	    Date inittime=new Date(System.currentTimeMillis());
	    if(starttime.after(inittime) ){
	    	delay = (starttime.getTime()-inittime.getTime());
	    }	
	    scheduler.schedule(event, delay, TimeUnit.MILLISECONDS);

	}
	
	public static void main(String[] args) throws InterruptedException, ParseException, IOException, ClassNotFoundException {
		Schedule schedule = new Schedule();
		
		FileInputStream fis = new FileInputStream("D:\\workspace\\task.dat");
		ObjectInputStream ois = new ObjectInputStream(fis);
		Task newtk = (Task)ois.readObject();
		
		schedule.repeat(new MyThread(Cycle.MINUTE, newtk), "2008/11/11 20:21:00");		
	}
}
/**
 * 导数据用线程
 * @author Administrator
 *
 */
class MyThread implements Runnable{
	
	private static int taskCount = 0;
	private final int id = taskCount++;
	Cycle period;
	Task newtk = null;
	
	public MyThread(Cycle period,Task newtk){
		this.period = period;
		this.newtk = newtk;
	}
	public void run() {
			System.out.println("t["+ id + "] is running. Now time "+
					new Date(System.currentTimeMillis()));
			
//          *****************************与源库建立连接，从源库取得数据*****************************
	        try {
				Class.forName("oracle.jdbc.driver.OracleDriver");
			} catch (ClassNotFoundException e) {
				System.out.println("无法完成数据库驱动！");
				e.printStackTrace();
			}
	        Connection conn = null;
			try {
				conn = DriverManager.getConnection(newtk.getSURL(),newtk.getSuser(),newtk.getSpassword());
			} catch (SQLException e) {
				System.out.println("无法与源数据库建立连接！");
				e.printStackTrace();
			}
	        
	        String field = "";
	        for(int i = 0; i < newtk.getSAttrb().length; i++){
	        	field =field +  "," + newtk.getSAttrb()[i];
	        }
	        field = field.substring(1);
	        String query = " select " + field + " from " + newtk.getSSchema() + "." + newtk.getSTab();
	        
	        PreparedStatement pstm = null;
	        ResultSet rs = null;
			try {
				pstm = conn.prepareStatement(query);
				rs = pstm.executeQuery();
			} catch (SQLException e) {
				System.out.println("无法从源数据库中取得数据！");
				e.printStackTrace();
			}
			
            //*****************************与目的库建立连接，向目的库数据*****************************
			 Connection newconn = null;
				try {
					newconn = DriverManager.getConnection(newtk.getDURL(),newtk.getDuser(),newtk.getDpassword());
				} catch (SQLException e) {
					System.out.println("无法与目的数据库建立连接！");
					e.printStackTrace();
				}
				
	        String newfield = "";
	        for(int i = 0; i < newtk.getDAttrb().length; i++){
	        	newfield = newfield + "," + newtk.getDAttrb()[i];
	        }
	        newfield = newfield.substring(1);
	        String att = "";
	        for(int i = 0;i < newtk.getDAttrb().length; i++){
	        	att = att + "," + "?";
	        }
	        att = att.substring(1);
	        String insert = "insert into " + newtk.getSSchema() + "." + newtk.getDTab() + "(" + newfield + ") values (" + att +")" ;
	        PreparedStatement inpstm = null;
		    System.out.println(insert);
	        int num = 0;
	        
	        try {
				while(rs.next()){
					
					inpstm = newconn.prepareStatement(insert);
					int attnum = 1;
					for(int i = 0; i < newtk.getDAttrb().length; i++){
		        		if(newtk.getD()[i].equalsIgnoreCase("int"))
		        			inpstm.setInt(attnum,rs.getInt(newtk.getDAttrb()[i]));
		        		if(newtk.getD()[i].equalsIgnoreCase("String"))
		        			inpstm.setString(attnum,rs.getString(newtk.getDAttrb()[i]));
		        		if(newtk.getD()[i].equalsIgnoreCase("Date"))
		        			inpstm.setDate(attnum,rs.getDate(newtk.getDAttrb()[i]));
		        		if(newtk.getD()[i].equalsIgnoreCase("float"))
		        			inpstm.setFloat(attnum,rs.getFloat(newtk.getDAttrb()[i]));
		        		if(newtk.getD()[i].equalsIgnoreCase("double"))
		        			inpstm.setDouble(attnum,rs.getDouble(newtk.getDAttrb()[i]));
		        		
		        		attnum++;
		        	}			
	
					num += inpstm.executeUpdate();

				}
			} catch (SQLException e) {
				System.out.println("向目的库数据库中导入数据失败！");
				e.printStackTrace();
				
			}
			
	        System.out.println("插入了  " + num + " 项数据！");

	        try {
				inpstm.close();
				newconn.close();
				
				pstm.close();
		        rs.close();
		        conn.close();
			
			} catch (SQLException e) {
				e.printStackTrace();
			}
//			*****************************周期运行线程，启动下一次任务*****************************		
			Date starttime = new Date(System.currentTimeMillis());
			Calendar c = new GregorianCalendar();
		    c.setTime(starttime);
		    switch(this.period){
		    case YEAR:
		    	c.add(Calendar.YEAR, 1);
		    	break;
		    case MONTH:
		    	c.add(Calendar.MONTH, 1);
		    	break;
		    case WEEK:
		    	c.add(Calendar.DATE, 7);
		    	break;
		    case DAY:
		    	c.add(Calendar.DATE, 1);
		    	break;
		    case HOUR:
		    	c.add(Calendar.HOUR, 1);
		    	break;
		    case MINUTE:
		    	c.add(Calendar.MINUTE, 1);
		    	break;
		    }
		    long delay = c.getTimeInMillis() - starttime.getTime();
		    Schedule.scheduler.schedule(new MyThread(this.period,newtk), delay, TimeUnit.MILLISECONDS);
		    
	}
}