schedule.java

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

package hdaty.prtds.rdgadaptor.schedule;

import hdaty.prtds.rdgadaptor.protocol.Message;

import java.io.File;
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;
import java.util.TimeZone;
import java.util.Vector;

/**
 * 线程执行的周期:<br>
 * 年、月、周、日、时、分
 * @author Administrator
 *
 */
	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 message  任务开始执行的时间
	 * @throws InterruptedException
	 * @throws ParseException
	 */
	public void repeat(Runnable event, Message message)
	throws InterruptedException, ParseException {

		long delay = 0;
		String date = message.getDATE()[0];
		System.out.println(message.getDATE()[0]);

		TimeZone tz = TimeZone.getTimeZone("GMT+8");
		Calendar now = Calendar.getInstance(tz);

		int years = now.get(Calendar.YEAR);
		int months = now.get(Calendar.MONTH);
		int days = now.get(Calendar.DAY_OF_MONTH);
		int weeks = now.get(Calendar.DAY_OF_WEEK);
		int hours = now.get(Calendar.HOUR_OF_DAY);
		int minutes = now.get(Calendar.MINUTE);
	//	int seconds = now.get(Calendar.SECOND);


		Date inittime = new Date(System.currentTimeMillis());// 任务启动时间		
		if (message.getPeriod().equals("每月")) {// DD:hh:mm 日期：小时：分钟
			int sDay = Integer.parseInt(date.split("号")[0]);
			String setedtime = null;
			setedtime = date.split(" ")[1];
			int sHour=Integer.parseInt(setedtime.substring(0, 2));			
			int sMinute= Integer.parseInt(setedtime.substring(3, 5));
			System.out.println(sHour);
			System.out.println(sMinute);
			now.set(years, months, sDay, sHour, sMinute, 0);//设定当月定时点的时间
			Date sysdate = new Date(now.getTimeInMillis());
			System.out.println(sysdate.toString());
		//	if (days>sDay&& hours>sHour&& minutes>sMinute) now.add(Calendar.MONTH, 1);// 启动时间大于定时时间
			if(days > sDay || (days == sDay && hours > sHour) || (days == sDay && hours == sHour && minutes > sMinute)){
				now.add(Calendar.MONTH, 1);
			}
		}

		if (message.getPeriod().equals("每天")) {
			int sHour=Integer.parseInt(date.substring(0, 2));
			int sMinute= Integer.parseInt(date.substring(3, 5));	
			now.set(years, months, days, sHour, sMinute, 0);//当天定时点的时间	
		//	if (hours> sHour && minutes>sMinute) now.add(Calendar.DATE, 1);// 启动时间小于定时时间
			System.out.println(hours);
			System.out.println(sHour);
			System.out.println(sMinute);
			if(hours > sHour ||(hours == sHour && minutes > sMinute)){
				now.add(Calendar.DATE, 1);
				System.out.println("在下个周期执行！");
			}
		}
	
	   if (message.getPeriod().equals("每周")) {// DD:HH:mm
		    String week = date.split(" ")[0];
		    int sWeek = 2;//默认每周一执行任务
		    if(week.equals("周日")){
		    	sWeek = 1;
		    }
		    else if(week.equals("周一")){
		    	sWeek = 2;
		    }
		    else if(week.equals("周二")){
		    	sWeek = 3;
		    }
		    else if(week.equals("周三")){
		    	sWeek = 4;
		    }
		    else if(week.equals("周四")){
		    	sWeek = 5;
		    }
		    else if(week.equals("周五")){
		    	sWeek = 6;
		    }
		    else if(week.equals("周六")){
		    	sWeek = 7;
		    }
		    String settime = null;
		    settime = date.split(" ")[1];
			int sHour=Integer.parseInt(settime.substring(0, 2));
			int sMinute= Integer.parseInt(settime.substring(3, 5));
			now.set(years, months, days, sHour, sMinute, 0);//当天定时点的时间
//			if ((weeks <= sWeek	&& hours <= sHour && minutes <= sMinute))  now.add(Calendar.DATE, sWeek-weeks);// 启动时间不大于定时时间
//		    else now.add(Calendar.DATE, sWeek-weeks+7);//下周定时点执行时间点	
			if(weeks < sWeek ||(weeks == sWeek && hours < sHour) || (weeks ==sWeek && hours == sHour && minutes < sMinute)){
				now.add(Calendar.DATE, sWeek-weeks);
			}
			else{
				now.add(Calendar.DATE, sWeek-weeks+7);
			}
		}
	   Date sysdate = new Date(now.getTimeInMillis());
	   System.out.println(sysdate.toString());
	   System.out.println(now.toString());

	   
	   delay = (now.getTimeInMillis() - inittime.getTime());
	   System.out.println(inittime.toString());
	   System.out.println(now.getTimeInMillis());
	   System.out.println(inittime.getTime());
	   System.out.println(delay);
	   scheduler.schedule(event, 0, TimeUnit.MILLISECONDS);
	}
	/**
	 * 测试用主函数
	 * @param args
	 * @throws InterruptedException
	 * @throws ParseException
	 * @throws IOException
	 * @throws ClassNotFoundException
	 */
	public static void main(String[] args) throws InterruptedException, ParseException, IOException, ClassNotFoundException {
		Schedule schedule = new Schedule();	
		
		FileInputStream fis2 = new FileInputStream("./src/数据包/task.txt");
		ObjectInputStream ois2 = new ObjectInputStream(fis2);
		Vector tasks = (Vector)ois2.readObject();
		System.out.println(tasks.size());
		for( int i = 0; i < tasks.size(); i ++){
			Message taskmessage = (Message) tasks.elementAt(i);
			System.out.println(taskmessage.getTaskName());
			if(taskmessage.isExcute()){
				Cycle period = Cycle.DAY ;
			    if(taskmessage.getPeriod().equals("每月"))
					period = Cycle.MONTH;
				else if(taskmessage.getPeriod().equals("每周"))
					period = Cycle.WEEK;
				else if(taskmessage.getPeriod().equals("每天"))
					period = Cycle.DAY;
				schedule.repeat(new MyThread(period,taskmessage), taskmessage);
				System.out.println(period);
			}
		}
	}
}

/**
 * 导数据用线程，该线程实现了，导数据及周期性的执行导数据
 * @author Administrator
 *
 */
class MyThread implements Runnable{
	
	String root = System.getProperty("user.dir");
	Log log = new Log();

	Date starttime = null;
	private static int taskCount = 0;
	private final int id = taskCount++;
	private Cycle period;
	private Message newtk = null;
	
	public MyThread(Cycle period,Message newtk){
		this.period = period;
		this.newtk = newtk;
	}
	
	/**
	 * 增量导数据模式
	 *
	 */
	public void increment(){
		    
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss:  ");
//		System.out.println(newtk.getDATE());
//		String logerrorname = null;
//		String logsqlname = null;
		
//		logerrorname = sdf.format(new Date())+ "_系统错误.log";
//		logsqlname = sdf.format(new Date()) + "_SQL.log";
		String logerrorname = null;
		String logsqlname = null;
		logerrorname = "log.txt";
		logsqlname = "log.txt";
		
		String error = "";
		String sql = "";
		boolean bFile;
		bFile = false;
		File dirFile;
		dirFile = new File("" + root + "/src/Log");
		bFile = dirFile.exists();

		if (bFile == true) {
			 System.out.println("Log文件夹存在。");
		} else {
			System.out.println("Log文件夹不存在" + "\n" +"创建Log文件夹"+ "\n");
			bFile = dirFile.mkdir();
			if (bFile == true) {
				System.out.println("Log文件夹 Createsuccessfully!");
			} else {
				 System.out.println("Disable to make the folder,please check the disk is full or not.");
				System.exit(1);
			}
		}
		
		System.out.println("开始执行增量模式导数据！………………");
		
//      *****************************与源库建立连接，从源库取得数据***************************************
        try {
			Class.forName(newtk.getSDriver());
		} catch (ClassNotFoundException e) {
			error += sdf.format(new Date()) + " 方案:" + newtk.getTaskName()+ " 无法完成源数据库驱动！\n";			
			log.Log("" + root + "/src/Log/" + logerrorname, error);
			System.out.println("无法完成源数据库驱动！");
			e.printStackTrace();
		}
        Connection conn = null;
		try {
			conn = DriverManager.getConnection(newtk.getSURL(),newtk.getSuser(),newtk.getSpassword());
		} catch (SQLException e) {
			error += sdf.format(new Date()) + " 方案:" + newtk.getTaskName()+ " 无法与源数据库建立连接！";
			log.Log("" + root + "/src/Log/" + logerrorname, error);
			System.out.println("无法与源数据库建立连接！");
			e.printStackTrace();
		}
        
        String field = "";
        for(int i = 0; i < newtk.getS().length; i++){
        	field =field +  "," + newtk.getS()[i];
        }
        field = field.substring(1);
//        newtk.getSKind().equals("");
        String sschema = null;
        if(newtk.getSSchema() == null){
        	sschema = "";
        }
        else{
        	sschema = newtk.getSSchema();
        }
        String stable = null;
        if(newtk.getSKind().equalsIgnoreCase("excel")){
        	stable = "[Sheet1$]";
        }
        else{
        	stable = newtk.getSTab();
        }
        String query = " select " + field + " from " + sschema + "." + stable;
        
        PreparedStatement pstm = null;
        ResultSet rs = null;
		try {
			pstm = conn.prepareStatement(query);
			rs = pstm.executeQuery();
		} catch (SQLException e) {
			error += sdf.format(new Date()) +" 方案:" + newtk.getTaskName()+  " 无法从源数据库中取得数据！";
			log.Log("" + root + "/src/Log/" + logerrorname, error);
			System.out.println("无法从源数据库中取得数据！");
			e.printStackTrace();
		}
		
//       *****************************与目的库建立连接，向目的库数据*************************************************
		try {
			Class.forName(newtk.getDDriver());
		} catch (ClassNotFoundException e) {
			error += sdf.format(new Date()) + " 方案:" + newtk.getTaskName()+ " 无法完成目的数据库驱动！";
			log.Log("" + root + "/src/Log/" + logerrorname, error);
			System.out.println("无法完成目的数据库驱动！");
			e.printStackTrace();
		} 
		Connection newconn = null;
			try {
				//DriverManager.registerDriver(new oracle.jdbc.OracleDriver()); 
				newconn = DriverManager.getConnection(newtk.getDURL(),newtk.getDuser(),newtk.getDpassword());
			} catch (SQLException e) {
				error += sdf.format(new Date()) + " 方案:" + newtk.getTaskName()+ " 无法与目的数据库建立连接！";
				log.Log("" + root + "/src/Log/" + logerrorname, error);
				System.out.println("无法与目的数据库建立连接！");
				e.printStackTrace();
			}
			
        String newfield = "";
        for(int i = 0; i < newtk.getD().length; i++){
        	newfield = newfield + "," + newtk.getD()[i];
        }
        newfield = newfield.substring(1);
        String att = "";
        for(int i = 0;i < newtk.getD().length; i++){