hadoop-1

结合其他文章

http://weixiaolu.iteye.com/blog/1504898

https://www.cnblogs.com/dycg/p/3934394.html

https://blog.csdn.net/c929833623lvcha/article/details/49052845

和自己看的，记录下

如何用？

实现 VersionedProtocol 定义方法
Server server = new RPC.Builder(conf).setProtocol(TestProtocol.class)
                .setInstance(new TestProtocolImpl()).setBindAddress(ADDRESS).setPort(0)
                .build();
server.start();
-----------------------------------------------------------------------------------------
TestProtocol proxy = RPC.getProxy(TestProtocol.class,TestProtocol.versionID,addr, conf);
proxy.调用定义的方法

如何实现？

Service:

先从启动看起，包含了Responder/Listener/Handler[]，他们分别处理Nio过程中的不同步骤

  public synchronized void start() {
    responder.start();
    listener.start();
    handlers = new Handler[handlerCount];
    
    for (int i = 0; i < handlerCount; i++) {
      handlers[i] = new Handler(i);
      handlers[i].start();
    }
  }

觉得别人写的不错，抄自：https://www.cnblogs.com/zhixingheyi2016/p/8781006.html

服务端使用了四组线程

Listener：

单个线程，用于监听连接，持有selector ，然后从 Reader[]线程组内挑选一个线程接受监听好的SelectionKey

// create a selector;
selector= Selector.open();
readers = new Reader[readThreads];
for (int i = 0; i < readThreads; i++) {
   Reader reader = new Reader(
     "Socket Reader #" + (i + 1) + " for port " + port);
   readers[i] = reader;
   reader.start();
}

Reader[]：

reader 线程组负责读取连接上的读请求，并传递给 Handler线程组，每个线程持有一个readSelector

 void doRead(SelectionKey key) throws InterruptedException {
      int count;
      Connection c = (Connection)key.attachment();
      ......try {
        //处理读
        count = c.readAndProcess();
      } catch (InterruptedException ieo) {
}

    private void processOneRpc(ByteBuffer bb)
        throws IOException, InterruptedException {try {
        ......
        callId = header.getCallId();
        ......final RpcCall call = new RpcCall(this, callId, retry);
        setupResponse(call,
            rse.getRpcStatusProto(), rse.getRpcErrorCodeProto(), null,
            t.getClass().getName(), t.getMessage());
        sendResponse(call);
      }
    }

Handler[]：

hander 线程组负责处理请求并返回响应，未能成功返回的响应交Responder

  private class Handler extends Thread {
    public Handler(int instanceNumber) {
      this.setDaemon(true);
      this.setName("IPC Server handler "+ instanceNumber + " on " + port);
    }

    @Override
    public void run() {
      LOG.debug(Thread.currentThread().getName() + ": starting");
      SERVER.set(Server.this);
      while (running) {
        TraceScope traceScope = null;
        try {
          final Call call = callQueue.take(); // pop the queue; maybe blocked here
          ...
          call.run();
        }
      }
    }
  }

Responder：

持有一个writeSelector，监听写事件

对于其中的Call对象：该类封装了一个RPC请求，它主要包含唯一标识id，函数调用信息、函数执行返回值value，异常信息error和执行完成标识done。由于HadoopRPCServer采用了异步方式处理客户端请求，这使得远程过程调用的发生顺序与结果返回顺序无直接关系，而Client端正是通过id识别不同的函数调用。当客户端向服务端发送请求时，只需要填充id和param这两个变量，而剩下的三个变量：value,error,done,则由服务端根据函数执行情况填充.

对于Connection, 用于Client与每个Server之间维护一个通信连接。该连接相关的基本信息及操作被封装到Connection类中，其中基本信息主要包括：通信连接唯一标识remoteId,与Server端通信的Socket,网络输入流in,网络输出流out,保存RPC请求的哈希表calls等.

public class Connection {
    private Socket socket;private LinkedList<RpcCall> responseQueue;
......

Invoker的invoke方法

        val = (RpcWritable.Buffer) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
            new RpcProtobufRequest(rpcRequestHeader, theRequest), remoteId,
            fallbackToSimpleAuth);

客户端设计

client.call

  Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
      ConnectionId remoteId, int serviceClass,
      AtomicBoolean fallbackToSimpleAuth) throws IOException {
    final Call call = createCall(rpcKind, rpcRequest);
    final Connection connection = getConnection(remoteId, call, serviceClass,
        fallbackToSimpleAuth);

    connection.sendRpcRequest(call);                 // send the rpc request

对于客户端和服务端的交互

Invoker ：动态代理，起始就是为了在invoke中实现具体的客户端访问逻辑，实现网络调用

    @Override
    public Object invoke(Object proxy, Method method, Object[] args)
      throws Throwable {
      try {
        value = (ObjectWritable)
          client.call(RPC.RpcKind.RPC_WRITABLE, new Invocation(method, args),
            remoteId, fallbackToSimpleAuth);
      } finally {
        if (traceScope != null) traceScope.close();
      }