将Hadoop RPC框架应用于多节点任务调度

/**
* 定时任务信息，包括任务信息和触发器信息
*/
public class CronJobInfo implements Writable
{
    private JobDetailInfo jobDetailInfo = new JobDetailInfo();
    private TriggerInfo triggerInfo = new TriggerInfo();
    @Override
    public void readFields(DataInput in) throws IOException
    {
        jobDetailInfo.readFields(in);
        triggerInfo.readFields(in);
    }
    @Override
    public void write(DataOutput out) throws IOException
    {
        jobDetailInfo.write(out);
        triggerInfo.write(out);
    }
    // getters and setters...
}

public class JobDetailInfo implements Writable
{
    private String name; // 任务名称
    private String group = Scheduler.DEFAULT_GROUP; // 任务组
    private String description; // 任务描述
    private Class jobClass; // 任务的启动类
    private JobDataMap jobDataMap; // 任务所需的参数，用来给作业提供数据支持的数据结构
    private boolean volatility = false; // 重启应用之后是否删除任务的相关信息,
    private boolean durability = false; // 任务完成之后是否依然保留到数据库
    private boolean shouldRecover = false; // 应用重启之后时候忽略过期任务
    @Override
    public void readFields(DataInput in) throws IOException
    {
        name = WritableUtils.readString(in);
        group = WritableUtils.readString(in);
        description = WritableUtils.readString(in);
        String className = WritableUtils.readString(in);
        if (className != null)
        {
          try
          {
             jobClass = Class.forName(new String(className));
          }
          catch (ClassNotFoundException e)
          {
             e.printStackTrace();
          }
        }
        int dataMapSize = WritableUtils.readVInt(in);
        while (dataMapSize-- > 0)
        {
           String key = WritableUtils.readString(in);
           String value = WritableUtils.readString(in);
           jobDataMap.put(key, value);
        }
        volatility = in.readBoolean();
        durability = in.readBoolean();
        shouldRecover = in.readBoolean();
    }
    @Override
    public void write(DataOutput out) throws IOException
    {
        WritableUtils.writeString(out, name);
        WritableUtils.writeString(out, group);
        WritableUtils.writeString(out, description);
        WritableUtils.writeString(out, jobClass.getName());
        if (jobDataMap == null)
            WritableUtils.writeVInt(out, 0);
        else
        {
            WritableUtils.writeVInt(out, jobDataMap.size());
            for (Object k : jobDataMap.keySet())
            {
                WritableUtils.writeString(out, k.toString());
                WritableUtils.writeString(out, jobDataMap.get(k).toString());
            }
        }
        out.writeBoolean(volatility);
        out.writeBoolean(durability);
        out.writeBoolean(shouldRecover);
   }
   //getters and setters
   //.....
}

public class TriggerInfo implements Writable
{
    private String name; // trigger名称
    private String group = Scheduler.DEFAULT_GROUP; // triger组名称
    private String description; // trigger描述
    private Date startTime; // 启动时间
    private Date endTime; // 结束时间
    private long repeatInterval; // 重试时间间隔
    private int repeatCount; //重试次数
    @Override
    public void readFields(DataInput in) throws IOException
    {
       name = WritableUtils.readString(in);
       group = WritableUtils.readString(in);
       description = WritableUtils.readString(in);
       long start = in.readLong();
       startTime = start==0 ? null : new Date(start);
       long end = in.readLong();
       endTime = end==0 ? null : new Date(end);
       repeatInterval = in.readLong();
       repeatCount = in.readInt();
    }
    @Override
    public void write(DataOutput out) throws IOException
    {
       WritableUtils.writeString(out, name);
       WritableUtils.writeString(out, group);
       WritableUtils.writeString(out, description);
       out.writeLong(startTime == null ? 0 : startTime.getTime());
       out.writeLong(endTime == null ? 0 : endTime.getTime());
       out.writeLong(repeatInterval);
       out.writeInt(repeatCount);
    }
    //getters and setters
    //.....
}

public interface TaskProtocol extends VersionedProtocol
{
	public CronJobInfo hearbeat();
}

public class TaskScheduler implements TaskProtocol
{
	private Logger logger = Logger.getLogger(getClass());
	private Server server;
	public TaskScheduler()
	{
		try
		{
			server = RPC.getServer(this, "192.168.1.101", 8888, new Configuration());
			server.start();
			server.join();
		}
		catch (UnknownHostException e)
		{
			e.printStackTrace();
		}
		catch (IOException e)
		{
			e.printStackTrace();
		}
		catch (InterruptedException e)
		{
			e.printStackTrace();
		}
	}
	@Override
	public long getProtocolVersion(String arg0, long arg1) throws IOException
	{
		return 1;
	}
	@Override
	public CronJobInfo generateCronJob()
	{
		// 1、创建JobDetial对象
		JobDetailInfo detail = new JobDetailInfo();
		// 设置工作项
		detail.setJobClass(DemoTask.class);
		detail.setName("MyJob_1");
		detail.setGroup("JobGroup_1");
		// 2、创建Trigger对象
		TriggerInfo trigger = new TriggerInfo();
		trigger.setName("Trigger_1");
		trigger.setGroup("Trigger_Group_1");
		trigger.setStartTime(new Date());
		// 设置重复停止时间，并销毁该Trigger对象
		Calendar c = Calendar.getInstance();
		c.setTimeInMillis(System.currentTimeMillis() + 1000 * 1L);
		trigger.setEndTime(c.getTime());
		// 设置重复间隔时间
		trigger.setRepeatInterval(1000 * 1L);
		// 设置重复执行次数
		trigger.setRepeatCount(3);
		CronJobInfo info = new CronJobInfo();
		info.setJobDetailInfo(detail);
		info.setTriggerInfo(trigger);
		return info;
	}
	public static void main(String[] args)
	{
		TaskScheduler ts = new TaskScheduler();
	}
}

public class DemoTask implements Job
{
	public void execute(JobExecutionContext context)
			throws JobExecutionException
	{
		System.out.println(this + ": executing task @" + new Date());
	}
}

public class TaskRunner
{
	private Logger logger = Logger.getLogger(getClass());
	private TaskProtocol proxy;
	public TaskRunner()
	{
		InetSocketAddress addr = new InetSocketAddress("localhost", 8888);
		try
		{
			proxy = (TaskProtocol) RPC.waitForProxy(TaskProtocol.class, 1, addr,
					new Configuration());
		}
		catch (IOException e)
		{
			e.printStackTrace();
		}
	}
	public void close()
	{
		RPC.stopProxy(proxy);
	}
	/**
	 * 从server获取一个定时任务
	 */
	public void getCronJob()
	{
		CronJobInfo info = proxy.generateCronJob();
		JobDetail jobDetail = getJobDetail(info.getJobDetailInfo());
		SimpleTrigger trigger = getTrigger(info.getTriggerInfo());
		// 创建Scheduler对象，并配置JobDetail和Trigger对象
		SchedulerFactory sf = new StdSchedulerFactory();
		Scheduler scheduler = null;
		try
		{
			scheduler = sf.getScheduler();
			scheduler.scheduleJob(jobDetail, trigger);
			// 执行启动操作
			scheduler.start();
		}
		catch (SchedulerException e)
		{
			e.printStackTrace();
		}
	}
	/**
	 * @param jobDetailInfo
	 * @return
	 */
	private JobDetail getJobDetail(JobDetailInfo info)
	{
		JobDetail detail = new JobDetail();
		detail.setName(info.getName());
		detail.setGroup(info.getGroup());
		detail.setDescription(info.getDescription());
		detail.setJobClass(info.getJobClass());
		detail.setJobDataMap(info.getJobDataMap());
		detail.setRequestsRecovery(info.isShouldRecover());
		detail.setDurability(info.isDurability());
		detail.setVolatility(info.isVolatility());
		logger.info("client get jobdetail:" + detail);
		return detail;
	}
	/**
	 * @param triggerInfo
	 * @return
	 */
	private SimpleTrigger getTrigger(TriggerInfo info)
	{
		SimpleTrigger trigger = new SimpleTrigger();
		trigger.setName(info.getName());
		trigger.setGroup(info.getGroup());
		trigger.setDescription(info.getDescription());
		trigger.setStartTime(info.getStartTime());
		trigger.setEndTime(info.getEndTime());
		trigger.setRepeatInterval(info.getRepeatInterval());
		trigger.setRepeatCount(info.getRepeatCount());
		logger.info("client get trigger:" + trigger);
		return trigger;
	}
	public static void main(String[] args)
	{
		TaskRunner t = new TaskRunner();
		t.getCronJob();
		t.close();
	}
}