|
原文链接:http://standalone.iteye.com/blog/1727544
现在版本的hadoop各种server、client RPC端通信协议的实现是基于google的protocol buffers的,如果对这个不熟悉,读code的时候会比较痛苦一些,所以花了些时间学习了一下,然后仿照写了个比较简单的例子,麻雀虽小,五脏俱全,看懂了我这个或许对你读hadoop的code有帮助! :)
我现在实现一个简单的server-client方式的calculator,client将计算请求序列化成protocol buffers形式然后发给server端,server端反序列化后将完成计算然后将结果序列化后返回给client端。
先看一下最后整体的package结构(模仿hadoop的包命名,便于比较)
package org.tao.pbtest.api:
org.tao.pbtest.api.Calculator
org.tao.pbtest.api.CalculatorPB
org.tao.pbtest.api.CalculatorPBServiceImpl
package org.tao.pbtest.server.business
org.tao.pbtest.server.business.CalculatorService
package org.tao.pbtest.ipc
org.tao.pbtest.ipc.Server
package org.tao.pbtest.proto
org.tao.pbtest.proto.Calculator
org.tao.pbtest.proto.CalculatorMsg
package org.tao.pbtest.proto.test
org.tao.pbtest.proto.test.TestCalculator
首先看一下Calculator这个接口:
Java代码
- package org.tao.pbtest.api;
-
- public interface Calculator {
- public int add(int a, int b);
- public int minus(int a, int b);
- }
package org.tao.pbtest.api;
public interface Calculator {
public int add(int a, int b);
public int minus(int a, int b);
}
这个计算器就进行简单的两种运算,两个整数的加减。
然后定义两个proto文件:CalculatorMsg.proto和Calculator.proto。
第一个是运算的参数消息、返回结果消息,输入时两个整数,返回结果是一个整数。具体protocol buffers的语法此处不做解释了,可以参看google的文档。
Java代码
- option java_package = "org.tao.pbtest.proto";
- option java_outer_classname = "CalculatorMsg";
- option java_generic_service = true;
- option java_generate_equals_and_hash = true;
-
- message RequestProto {
- required string methodName = 1;
- required int32 num1 = 2;
- required int32 num2 = 3;
- }
-
- message ResponseProto {
- required int32 result = 1;
- }
option java_package = "org.tao.pbtest.proto";
option java_outer_classname = "CalculatorMsg";
option java_generic_service = true;
option java_generate_equals_and_hash = true;
message RequestProto {
required string methodName = 1;
required int32 num1 = 2;
required int32 num2 = 3;
}
message ResponseProto {
required int32 result = 1;
}
第二个proto文件定义service:
Java代码
- option java_package = "org.tao.pbtest.proto";
- option java_outer_classname = "Calculator";
- option java_generic_service = true;
- option java_generate_equals_and_hash = true;
-
- import "CalculatorMsg.proto"
-
- service CalculatorService {
- rpc add(RequestProto) returns (ResponseProto);
- rpc minus(RequestProto) returns (ResponseProto);
- }
option java_package = "org.tao.pbtest.proto";
option java_outer_classname = "Calculator";
option java_generic_service = true;
option java_generate_equals_and_hash = true;
import "CalculatorMsg.proto"
service CalculatorService {
rpc add(RequestProto) returns (ResponseProto);
rpc minus(RequestProto) returns (ResponseProto);
}
然后用protoc将此两个文件编译,生成两个java文件:
org.tao.pbtest.proto.Calculator
org.tao.pbtest.proto.CalculatorMsg
然后定义一个CalculatorPB接口extends刚才生成的org.tao.pbtest.proto.Calculator.CalculatorService.BlockingInterface, 这是一个过渡作用的接口。
Java代码
- package org.tao.pbtest.server.api;
-
- import org.tao.pbtest.proto.Calculator.CalculatorService.BlockingService;
-
- public interface CalculatorPB extends BlockingInterface {
- }
package org.tao.pbtest.server.api;
import org.tao.pbtest.proto.Calculator.CalculatorService.BlockingService;
public interface CalculatorPB extends BlockingInterface {
}
还需要一个发送、接受信息的ipc server/client端。这里偷懒只实现一个最最简单的server端,什么并发啊,异常处理啊,nio啊统统不考虑,因为这不是重点。
Java代码
- package org.tao.pbtest.ipc;
-
- import java.io.DataInputStream;
- import java.io.DataOutputStream;
- import java.io.IOException;
- import java.net.*;
- import com.google.protobuf.*;
- import com.google.protobuf.Descriptors.MethodDescriptor;
- import com.tao.pbtest.proto.CalculatorMsg.RequestProto;
- import com.tao.pbtest.proto.CalculatorMsg.ResponseProto;
-
- public class Server extends Thread {
- private Class protocol;
- private BlockingService impl;
- private int port;
- private ServerSocket ss;
-
- public Server(Class protocol, BlockingService protocolImpl, int port){
- this.protocol = protocol;
- this.impl = protocolImpl;
- this.port = port;
- }
-
- public void run(){
- Socket clientSocket = null;
- DataOutputStream dos = null;
- DataInputStream dis = null;
- try {
- ss = new ServerSocket(port);
- }catch(IOException e){
- }
- int testCount = 10; //进行10次计算后就退出
-
- while(testCount-- > 0){
- try {
- clientSocket = ss.accept();
- dos = new DataOutputStream(clientSocket.getOutputStream());
- dis = new DataInputStream(clientSocket.getInputStream());
- int dataLen = dis.readInt();
- byte[] dataBuffer = new byte[dataLen];
- byte[] result = processOneRpc(dataBuffer);
- dos.writeInt(result.length);
- dos.write(result);
- dos.flush();
- }catch(Exception e){
- }
- }
- try {
- dos.close();
- dis.close();
- ss.close();
- }catch(Exception e){
- };
-
- }
-
- public byte[] processOneRpc (byte[] data) throws Exception {
- RequestProto request = RequestProto.parseFrom(data);
- String methodName = request.getMethodName();
- MethodDescriptor methodDescriptor = impl.getDescriptorForType().findMethodByName(methodName);
- Message response = impl.callBlockingMethod(methodDescriptor, null, request);
- return response.toByteArray();
- }
- }
package org.tao.pbtest.ipc;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.*;
import com.google.protobuf.*;
import com.google.protobuf.Descriptors.MethodDescriptor;
import com.tao.pbtest.proto.CalculatorMsg.RequestProto;
import com.tao.pbtest.proto.CalculatorMsg.ResponseProto;
public class Server extends Thread {
private Class protocol;
private BlockingService impl;
private int port;
private ServerSocket ss;
public Server(Class protocol, BlockingService protocolImpl, int port){
this.protocol = protocol;
this.impl = protocolImpl;
this.port = port;
}
public void run(){
Socket clientSocket = null;
DataOutputStream dos = null;
DataInputStream dis = null;
try {
ss = new ServerSocket(port);
}catch(IOException e){
}
int testCount = 10; //进行10次计算后就退出
while(testCount-- > 0){
try {
clientSocket = ss.accept();
dos = new DataOutputStream(clientSocket.getOutputStream());
dis = new DataInputStream(clientSocket.getInputStream());
int dataLen = dis.readInt();
byte[] dataBuffer = new byte[dataLen];
byte[] result = processOneRpc(dataBuffer);
dos.writeInt(result.length);
dos.write(result);
dos.flush();
}catch(Exception e){
}
}
try {
dos.close();
dis.close();
ss.close();
}catch(Exception e){
};
}
public byte[] processOneRpc (byte[] data) throws Exception {
RequestProto request = RequestProto.parseFrom(data);
String methodName = request.getMethodName();
MethodDescriptor methodDescriptor = impl.getDescriptorForType().findMethodByName(methodName);
Message response = impl.callBlockingMethod(methodDescriptor, null, request);
return response.toByteArray();
}
}
CalculatorServer.java,实现计算器服务的类,此类依赖ipc Server接受请求并处理计算请求,注意到其自身实现了Calculator接口,本质上的计算是由其来完成的。也就是,Server接受客户端请求要执行方法M,Server对象里有实现了CalculatorPB接口的对象A,那么请求就交给A处理(A其实是CalculatorPBServiceImpl类的对象,此类后面介绍),此时A对应的M方法的参数是pb的形式,另外A对象里其实包含对CalculatorService的一个引用,所以在A的M方法里,先对参数反序列化,然后将参数交给CalculatorService处理。
Java代码
- package org.tao.pbtest.server.business;
-
- import java.lang.reflect.Constructor;
- import java.lang.reflect.InvocationTargetException;
- import java.lang.reflect.Method;
-
- import org.tao.pbtest.ipc.Server;
- import org.tao.pbtest.server.api.Calculator;
-
- import com.google.protobuf.BlockingService;
-
- public class CalculatorService implements Calculator {
-
- private Server server = null;
- private final Class protocol = Calculator.class;
- private final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
- private final String protoPackage = "org.tao.pbtest.proto";
- private final String host = "localhost";
- private final int port = 8038;
-
- public CalculatorService (){
-
- }
-
- @Override
- public int add(int a, int b) {
- // TODO Auto-generated method stub
- return a+b;
- }
-
- public int minus(int a, int b){
- return a-b;
- }
-
-
- public void init(){
- createServer();
- }
-
-
- /*
- * return org.tao.pbtest.server.api.CalculatorPBServiceImpl
- */
- public Class getPbServiceImplClass(){
- String packageName = protocol.getPackage().getName();
- String className = protocol.getSimpleName();
- String pbServiceImplName = packageName + "." + className + "PBServiceImpl";
- Class clazz = null;
- try{
- clazz = Class.forName(pbServiceImplName, true, classLoader);
- }catch(ClassNotFoundException e){
- System.err.println(e.toString());
- }
- return clazz;
- }
-
- /*
- * return org.tao.pbtest.proto.Calculator$CalculatorService
- */
- public Class getProtoClass(){
- String className = protocol.getSimpleName();
- String protoClazzName = protoPackage + "." + className + "$" + className + "Service";
- Class clazz = null;
- try{
- clazz = Class.forName(protoClazzName, true, classLoader);
- }catch(ClassNotFoundException e){
- System.err.println(e.toString());
- }
- return clazz;
- }
-
- public void createServer(){
- Class pbServiceImpl = getPbServiceImplClass();
- Constructor constructor = null;
- try{
- constructor = pbServiceImpl.getConstructor(protocol);
- constructor.setAccessible(true);
- }catch(NoSuchMethodException e){
- System.err.print(e.toString());
- }
-
- Object service = null; // instance of CalculatorPBServiceImpl
- try {
- service = constructor.newInstance(this);
- }catch(InstantiationException e){
- } catch (IllegalArgumentException e) {
- } catch (IllegalAccessException e) {
- } catch (InvocationTargetException e) {
- }
-
- /*
- * interface: org.tao.pbtest.server.CalculatorPB
- */
- Class pbProtocol = service.getClass().getInterfaces()[0];
-
- /*
- * class: org.tao.pbtest.proto.Calculator$CalculatorService
- */
- Class protoClazz = getProtoClass();
-
- Method method = null;
- try {
-
- // pbProtocol.getInterfaces()[] 即是接口 org.tao.pbtest.proto.Calculator$CalculatorService$BlockingInterface
-
- method = protoClazz.getMethod("newReflectiveBlockingService", pbProtocol.getInterfaces()[0]);
- method.setAccessible(true);
- }catch(NoSuchMethodException e){
- System.err.print(e.toString());
- }
-
- try{
- createServer(pbProtocol, (BlockingService)method.invoke(null, service));
- }catch(InvocationTargetException e){
- } catch (IllegalArgumentException e) {
- } catch (IllegalAccessException e) {
- }
-
- }
-
- public void createServer(Class pbProtocol, BlockingService service){
- server = new Server(pbProtocol, service, port);
- server.start();
- }
-
- public static void main(String[] args){
- CalculatorService cs = new CalculatorService();
- cs.init();
- }
- }
package org.tao.pbtest.server.business;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import org.tao.pbtest.ipc.Server;
import org.tao.pbtest.server.api.Calculator;
import com.google.protobuf.BlockingService;
public class CalculatorService implements Calculator {
private Server server = null;
private final Class protocol = Calculator.class;
private final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
private final String protoPackage = "org.tao.pbtest.proto";
private final String host = "localhost";
private final int port = 8038;
public CalculatorService (){
}
@Override
public int add(int a, int b) {
// TODO Auto-generated method stub
return a+b;
}
public int minus(int a, int b){
return a-b;
}
public void init(){
createServer();
}
/*
* return org.tao.pbtest.server.api.CalculatorPBServiceImpl
*/
public Class getPbServiceImplClass(){
String packageName = protocol.getPackage().getName();
String className = protocol.getSimpleName();
String pbServiceImplName = packageName + "." + className + "PBServiceImpl";
Class clazz = null;
try{
clazz = Class.forName(pbServiceImplName, true, classLoader);
}catch(ClassNotFoundException e){
System.err.println(e.toString());
}
return clazz;
}
/*
* return org.tao.pbtest.proto.Calculator$CalculatorService
*/
public Class getProtoClass(){
String className = protocol.getSimpleName();
String protoClazzName = protoPackage + "." + className + "$" + className + "Service";
Class clazz = null;
try{
clazz = Class.forName(protoClazzName, true, classLoader);
}catch(ClassNotFoundException e){
System.err.println(e.toString());
}
return clazz;
}
public void createServer(){
Class pbServiceImpl = getPbServiceImplClass();
Constructor constructor = null;
try{
constructor = pbServiceImpl.getConstructor(protocol);
constructor.setAccessible(true);
}catch(NoSuchMethodException e){
System.err.print(e.toString());
}
Object service = null; // instance of CalculatorPBServiceImpl
try {
service = constructor.newInstance(this);
}catch(InstantiationException e){
} catch (IllegalArgumentException e) {
} catch (IllegalAccessException e) {
} catch (InvocationTargetException e) {
}
/*
* interface: org.tao.pbtest.server.CalculatorPB
*/
Class pbProtocol = service.getClass().getInterfaces()[0];
/*
* class: org.tao.pbtest.proto.Calculator$CalculatorService
*/
Class protoClazz = getProtoClass();
Method method = null;
try {
// pbProtocol.getInterfaces()[] 即是接口 org.tao.pbtest.proto.Calculator$CalculatorService$BlockingInterface
method = protoClazz.getMethod("newReflectiveBlockingService", pbProtocol.getInterfaces()[0]);
method.setAccessible(true);
}catch(NoSuchMethodException e){
System.err.print(e.toString());
}
try{
createServer(pbProtocol, (BlockingService)method.invoke(null, service));
}catch(InvocationTargetException e){
} catch (IllegalArgumentException e) {
} catch (IllegalAccessException e) {
}
}
public void createServer(Class pbProtocol, BlockingService service){
server = new Server(pbProtocol, service, port);
server.start();
}
public static void main(String[] args){
CalculatorService cs = new CalculatorService();
cs.init();
}
}
刚才提到的PB格式跟最终实现的桥梁类:CalculatorPBServiceImpl
Java代码
- package org.tao.pbtest.server.api;
-
- import org.tao.pbtest.proto.CalculatorMsg.RequestProto;
- import org.tao.pbtest.proto.CalculatorMsg.ResponseProto;
-
- import com.google.protobuf.RpcController;
- import com.google.protobuf.ServiceException;
-
- public class CalculatorPBServiceImpl implements CalculatorPB {
-
- public Calculator real;
-
- public CalculatorPBServiceImpl(Calculator impl){
- this.real = impl;
- }
-
- @Override
- public ResponseProto add(RpcController controller, RequestProto request) throws ServiceException {
- // TODO Auto-generated method stub
- ResponseProto proto = ResponseProto.getDefaultInstance();
- ResponseProto.Builder build = ResponseProto.newBuilder();
- int add1 = request.getNum1();
- int add2 = request.getNum2();
- int sum = real.add(add1, add2);
- ResponseProto result = null;
- build.setResult(sum);
- result = build.build();
- return result;
- }
-
- @Override
- public ResponseProto minus(RpcController controller, RequestProto request) throws ServiceException {
- // TODO Auto-generated method stub
- ResponseProto proto = ResponseProto.getDefaultInstance();
- ResponseProto.Builder build = ResponseProto.newBuilder();
- int add1 = request.getNum1();
- int add2 = request.getNum2();
- int sum = real.minus(add1, add2);
- ResponseProto result = null;
- build.setResult(sum);
- result = build.build();
- return result;
- }
-
- }
package org.tao.pbtest.server.api;
import org.tao.pbtest.proto.CalculatorMsg.RequestProto;
import org.tao.pbtest.proto.CalculatorMsg.ResponseProto;
import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
public class CalculatorPBServiceImpl implements CalculatorPB {
public Calculator real;
public CalculatorPBServiceImpl(Calculator impl){
this.real = impl;
}
@Override
public ResponseProto add(RpcController controller, RequestProto request) throws ServiceException {
// TODO Auto-generated method stub
ResponseProto proto = ResponseProto.getDefaultInstance();
ResponseProto.Builder build = ResponseProto.newBuilder();
int add1 = request.getNum1();
int add2 = request.getNum2();
int sum = real.add(add1, add2);
ResponseProto result = null;
build.setResult(sum);
result = build.build();
return result;
}
@Override
public ResponseProto minus(RpcController controller, RequestProto request) throws ServiceException {
// TODO Auto-generated method stub
ResponseProto proto = ResponseProto.getDefaultInstance();
ResponseProto.Builder build = ResponseProto.newBuilder();
int add1 = request.getNum1();
int add2 = request.getNum2();
int sum = real.minus(add1, add2);
ResponseProto result = null;
build.setResult(sum);
result = build.build();
return result;
}
}
最后,偷懒没写客户端的东西,只是写了一个简单的测试例子:
Java代码
- package org.tao.pbtest.proto.test;
-
- import java.io.DataInputStream;
- import java.io.DataOutputStream;
- import java.io.IOException;
- import java.net.Socket;
- import java.util.Random;
-
- import org.tao.pbtest.proto.CalculatorMsg.RequestProto;
- import org.tao.pbtest.proto.CalculatorMsg.ResponseProto;
- import org.tao.pbtest.server.api.Calculator;
-
- public class TestCalculator implements Calculator {
-
- public int doTest(String op, int a, int b){
- // TODO Auto-generated method stub
- Socket s = null;
- DataOutputStream out = null;
- DataInputStream in = null;
- int ret = 0;
- try {
- s= new Socket("localhost", 8038);
- out = new DataOutputStream(s.getOutputStream());
- in = new DataInputStream(s.getInputStream());
-
- RequestProto.Builder builder = RequestProto.newBuilder();
- builder.setMethodName(op);
- builder.setNum1(a);
- builder.setNum2(b);
- RequestProto request = builder.build();
-
- byte [] bytes = request.toByteArray();
- out.writeInt(bytes.length);
- out.write(bytes);
- out.flush();
-
- int dataLen = in.readInt();
- byte[] data = new byte[dataLen];
- int count = in.read(data);
- if(count != dataLen){
- System.err.println("something bad happened!");
- }
-
- ResponseProto result = ResponseProto.parseFrom(data);
- System.out.println(a + " " + op + " " + b + "=" + result.getResult());
- ret = result.getResult();
-
- }catch(Exception e){
- e.printStackTrace();
- System.err.println(e.toString());
- }finally {
- try{
- in.close();
- out.close();
- s.close();
- }catch(IOException e){
- e.printStackTrace();
- }
- }
- return ret;
- }
- @Override
- public int add(int a, int b) {
- // TODO Auto-generated method stub
- return doTest("add", a, b);
- }
-
- @Override
- public int minus(int a, int b) {
- // TODO Auto-generated method stub
- return doTest("minus", a, b);
- }
-
- /**
- * @param args
- */
- public static void main(String[] args) {
- // TODO Auto-generated method stub
- TestCalculator tc = new TestCalculator();
- int testCount = 5;
- Random rand = new Random();
- while(testCount-- > 0){
- int a = rand.nextInt(100);
- int b = rand.nextInt(100);
- tc.add(a,b);
- tc.minus(a, b);
- }
-
- }
-
- }
package org.tao.pbtest.proto.test;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.Random;
import org.tao.pbtest.proto.CalculatorMsg.RequestProto;
import org.tao.pbtest.proto.CalculatorMsg.ResponseProto;
import org.tao.pbtest.server.api.Calculator;
public class TestCalculator implements Calculator {
public int doTest(String op, int a, int b){
// TODO Auto-generated method stub
Socket s = null;
DataOutputStream out = null;
DataInputStream in = null;
int ret = 0;
try {
s= new Socket("localhost", 8038);
out = new DataOutputStream(s.getOutputStream());
in = new DataInputStream(s.getInputStream());
RequestProto.Builder builder = RequestProto.newBuilder();
builder.setMethodName(op);
builder.setNum1(a);
builder.setNum2(b);
RequestProto request = builder.build();
byte [] bytes = request.toByteArray();
out.writeInt(bytes.length);
out.write(bytes);
out.flush();
int dataLen = in.readInt();
byte[] data = new byte[dataLen];
int count = in.read(data);
if(count != dataLen){
System.err.println("something bad happened!");
}
ResponseProto result = ResponseProto.parseFrom(data);
System.out.println(a + " " + op + " " + b + "=" + result.getResult());
ret = result.getResult();
}catch(Exception e){
e.printStackTrace();
System.err.println(e.toString());
}finally {
try{
in.close();
out.close();
s.close();
}catch(IOException e){
e.printStackTrace();
}
}
return ret;
}
@Override
public int add(int a, int b) {
// TODO Auto-generated method stub
return doTest("add", a, b);
}
@Override
public int minus(int a, int b) {
// TODO Auto-generated method stub
return doTest("minus", a, b);
}
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
TestCalculator tc = new TestCalculator();
int testCount = 5;
Random rand = new Random();
while(testCount-- > 0){
int a = rand.nextInt(100);
int b = rand.nextInt(100);
tc.add(a,b);
tc.minus(a, b);
}
}
}
输出:
76 add 14=90
76 minus 14=62
20 add 84=104
20 minus 84=-64
4 add 16=20
4 minus 16=-12
56 add 4=60
56 minus 4=52
46 add 50=96
46 minus 50=-4 |
|