HAMA 블로그

... 작성중 ...

...작성중..

hadoop-2.2.0 경우는 이클립스에서  maven  으로 import 하면 됩니다만

public class TestServer implements RPCProtocol {

Class  cls  = Class.forName(“RPCProtocol“);

Class paramTypes[] = new Class[2];
pramTypes[0] = String.TYPE;
pramTypes[2] = Integer.TYPE;

Method meth = cls.getMethod(“heartbeat”, paramTypes);

Object arglist[] = new Object[2];
arglist[0] = new String(“hello world”)
arglist[1] = new Integer(2);


RPCProtocol rp  =  new  TestServer ();

Object retObj = meth.invoke(rp, arglist);

Integer retval = (Integer) retObj;

중요포인트 ( 필요사항 ) :    실제 객체 /  클래스타입 /  메소드이름 /  인자타입 / 실제 인자 /    

Hadoop  에서 각각에 대한 변수명 

                     instance (실제객체) 
                     protocol   (클래스타입) 
                     Invoocation  ( 메소드이름, 인자클래스, 실제 인자값)

public class MyDebugInvocationHandler   implements java.lang.reflect.InvocationHandler {    
        private Object target = null;
          public void setTarget(Object target_) {
            this.target = target_;
        }
        
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                System.out.println("Going to execute method : " + method.getName);
                Object retObject = method.invoke(target, args);
                System.out.println("After execute method : " + method.getName());
                return retObject;
        }
    }


-----------


 IMyBusinessObject bo = new MyBusinessObject(); // 실제 비즈니스 객체
  MyDebugInvocationHandler aMyDebugInvocationHandler = new MyDebugInvocationHandler();
  aMyDebugInvocationHandler.setTarget(bo);
    
 IMyBusinessObject proxyObject =    (IMyBusinessObject) Proxy.newProxyInstance  // 프록시 객체 생성
          (IMyBusinessObject.class.getClassLoader(),
          new Class[] { IMyBusinessObject.class }, // 프록시할 인터페이스
          aMyDebugInvocationHandler); // InvocationHandler
    
    // 프록시 객체 사용
    System.out.println(proxyObject.doExecute("Hello World"));

* InputStream / OutputStream
* Selctor
* Channel
* ByteBuffer
* Socket

DFSClient

(ClientProtocol)RPC.getProxy(ClientProtocol.class,
        ClientProtocol.versionID, nameNodeAddr, ugi, conf,
        NetUtils.getSocketFactory(conf, ClientProtocol.class), 0,..)

 this.namenode = createNamenode(this.rpcNamenode, conf);
 
 namenode.getBlockLocations(src, start, length);

1)
   final Invoker invoker = new Invoker(protocol, addr, ticket, conf, factory,
        rpcTimeout, connectionRetryPolicy);
    VersionedProtocol proxy = (VersionedProtocol)Proxy.newProxyInstance(
        protocol.getClassLoader(), new Class[]{protocol}, invoker);

2)

  private static class Invoker implements InvocationHandler {
    private Client.ConnectionId remoteId;
    private Client client;
   
    private Invoker(Class<? extends VersionedProtocol> protocol,
                     InetSocketAddress address, 
                     UserGroupInformation ticket,
                     Configuration conf, 
                     SocketFactory factory,
                     int rpcTimeout, 
                     RetryPolicy connectionRetryPolicy) throws IOException {
   
       this.remoteId = Client.ConnectionId.getConnectionId(address, protocol,
          ticket, rpcTimeout, connectionRetryPolicy, conf);
       this.client = CLIENTS.getClient(conf, factory);
    }

    public Object invoke(Object proxy, Method method, Object[] args)
      throws Throwable {
    
      ObjectWritable value = (ObjectWritable)
        client.call(new Invocation(method, args), remoteId);
     
      return value.get();
    }

public Writable call(Writable param, ConnectionId remoteId)  
                       throws InterruptedException, IOException {
    Call call = new Call(param);
    Connection connection = getConnection(remoteId, call);
    connection.sendParam(call);                 // send the parameter

public class NameNode implements ClientProtocol, DatanodeProtocol,

   this.server = RPC.getServer(this, socAddr.getHostName(),
        socAddr.getPort(), handlerCount, false, conf, namesystem
        .getDelegationTokenSecretManager());

public static Server getServer(...){    
      return new Server(instance, conf, bindAddress, port, numHandlers, verbose, secretManager);
  }

 public static class Server extends org.apache.hadoop.ipc.Server {

  public Writable call(Class<?> protocol, Writable param, long receivedTime) 
        Invocation call = (Invocation)param;
        if (verbose) log("Call: " + call);

        Method method =
          protocol.getMethod(call.getMethodName(),
                                   call.getParameterClasses());
        method.setAccessible(true);

        long startTime = System.currentTimeMillis();
        Object value = method.invoke(instance, call.getParameters());

 Server
   listener = new Listener();
  
 Listener 
      void doAccept(SelectionKey key) 

  Reader 
    void doRead(SelectionKey key) 

  Connection 
         public int readAndProcess() 
         private void processData(byte[] buf) 
   
  Call 
  call = new Call(id, param, this); 
  callQueue.put(call); // queue the call; 

  Handler 
    final Call call = callQueue.take(); // pop the queue; 
    value = call(call.connection.protocol, call.param, call.timestamp); 

 Response

 Invocation call = (Invocation)param;
        
 Method method = protocol.getMethod(call.getMethodName(),call.getParameterClasses());
      
 Object value = method.invoke(instance, call.getParameters());