Hadoop 自定义RPC protocol

import org.apache.hadoop.ipc.VersionedProtocol;

public interface Protocol extends VersionedProtocol{
	public static final long versionID = 1L;
	public boolean writeFile(Info statics);

}

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;

public class Info implements Writable {
	public Text fileName;
	public LongWritable taskNum;
	public Info(){
		this.fileName=new Text();
		this.taskNum =new LongWritable();
		
		
	}
	public void setFileName(String str){
		this.fileName = new Text(str);
	}
	public void setTaskNum(Long num){
		this.taskNum = new LongWritable(num);
	}
	public String getFileName(){
		return fileName.toString();
	}
	public Long getTaskNum(){
		return taskNum.get();
	}
	@Override
	public void write(DataOutput out) throws IOException {
		fileName.write(out);
		taskNum.write(out);
		
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		System.out.println("Come to invoke my readFields");
		// TODO Auto-generated method stub
		fileName.readFields(in);
		taskNum.readFields(in);
		System.out.println("Read Success!!!");
	}

}

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;

import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.ProtocolSignature;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;



public class WriteServer implements Protocol{
	public Server server;

	@Override
	public long getProtocolVersion(String protocol, long clientVersion)
			throws IOException {
		return this.versionID;
	}

	@Override
	public ProtocolSignature getProtocolSignature(String protocol,
			long clientVersion, int clientMethodsHash) throws IOException {
		return new ProtocolSignature(Protocol.versionID,null);
	}

	@Override
	public boolean writeFile(Info statics) {
		// 在服务器端新建一个文件，并将由客户端传递过来的Info statics信息写入进去
		FileWriter writer;
		try {
			writer = new FileWriter("/home/chenershuile/helloWritable");
			BufferedWriter bw = new BufferedWriter(writer);
			bw.write(statics.getFileName());
			bw.write(statics.getTaskNum().toString());
			bw.close();
	    	writer.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return true;
// 执行完成，返回ture
	}



    public void init(){
    	try {
			this.server = new RPC.Builder(new Configuration()).setBindAddress("master").setNumHandlers(5).setProtocol(Protocol.class).setPort(50071)
					.setInstance(new WriteServer()).build();
		} catch (HadoopIllegalArgumentException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
    @SuppressWarnings("static-access")
	public void run(){
    	this.server.start();
    	System.out.println("The server start at:"+this.server.getPort());
    }
    public void stop(){
    	this.server.stop();
    }
}

public class App 
{
    public static void main( String[] args )
    {
    	WriteServer writeServer = new WriteServer();
    	writeServer.init();
    	writeServer.run();
    	
    }
}

import java.io.IOException;
import java.net.InetSocketAddress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;

import SJTU.server.Protocol;
import SJTU.server.Info;

/**
 * Hello world!
 *
 */
public class App 
{
    public static void main( String[] args )
    {
    	InetSocketAddress addr = new InetSocketAddress("master",50071);
    	Configuration conf = new Configuration();
        try {
			Protocol proxy=RPC.getProxy(Protocol.class,Protocol.versionID,addr,conf);
			System.out.println("Start Client");
			String fileName = "hdfs://";
			Long taskNum = 10L;
			Info statics = new Info();
			statics.setFileName(fileName);
			statics.setTaskNum(taskNum);
			System.out.println(proxy.writeFile(statics));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
}