HBase 中 HMaster、HRegionServer 和 Client 之間的通信使用了兩個(gè)技術(shù)区赵，Google Protobuf RPC 和 Java NIO郁副。

主要代碼位置：

.
|—— hbase-client
|    |—— org.apache.hadoop.hbase.ipc
|    |—— org.apache.hadoop.hbase.client
|—— hbase-server
|    |—— org.apache.hadoop.hbase.ipc
|    |—— org.apache.hadoop.hbase.master
|    |—— org.apache.hadoop.hbase.regionserver
|—— hbase-protocal 
|    |—— org.apache.hadoop.hbase.protobuf.generated
...

HBase RPC

什么是 PRC

RPC（Remote Procedure Call）即遠(yuǎn)程過程調(diào)用讨盒。對于本地調(diào)用，定義好一個(gè)函數(shù)以后愈案，程序的其他部分通過調(diào)用該函數(shù)挺尾，就可以返回想要的結(jié)果。RPC 的區(qū)別就是函數(shù)定義和函數(shù)調(diào)用位于不同的機(jī)器（大部分情況）站绪，因?yàn)樯婕暗讲煌臋C(jī)器遭铺，所以 RPC 相比較本地函數(shù)調(diào)用多了通信部分。主要涉及到兩個(gè)角色：調(diào)用方（client）和函數(shù)定義實(shí)現(xiàn)（server），RPC 調(diào)用的流程如下面圖所示：

RPC

HBase Server 端主要類關(guān)系：

[圖片上傳失敗...(image-6f67b9-1560062040880)]

HMaster 繼承 HRegionServer魂挂，rpcService 提供 RPC Server 端實(shí)現(xiàn)（HRegionServier 中由 RSRpcService 實(shí)現(xiàn)甫题，HMaster 中由 MasterRpcService 實(shí)現(xiàn)），rpcServer 是具體的 RPC Server 實(shí)現(xiàn)（實(shí)現(xiàn) RpcServerInterface 接口）涂召，Listener 線程負(fù)責(zé)監(jiān)聽請求坠非，Resonder 線程負(fù)責(zé)發(fā)送請求結(jié)果。

HBase Client 端主要類關(guān)系：

HBase Client

HBase Client 訪問 HBase 需要先創(chuàng)建 HConnection果正，Connection 中的 rpcClient（RpcClient 接口炎码，RpcClientImpl 是實(shí)現(xiàn)類）表示 Rpc Client 端實(shí)現(xiàn)，由 RpcClientFactory 創(chuàng)建秋泳。

RPC 初始化

在 HRegionServer 啟動類的源碼中潦闲，有以下代碼，分別初始化 RpcServer 和 RpcClient：

public HRegionServer(Configuration conf, CoordinatedStateManager csm)
      throws IOException, InterruptedException {
    // ...
    
    rpcServices = createRpcServices();
    
    // ...
}

private void initializeThreads() {
    // ...
    
    rpcClient = RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress(
        rpcServices.isa.getAddress(), 0), clusterConnection.getConnectionMetrics());    
    
    // ...
}

RPC Server

先來看以下 RPC Server 端的一些實(shí)現(xiàn)轮锥，從一些重要類的初始化開始

初始化

createRpcServices()

createRpcServices() 方法是初始化 RPC Server 端實(shí)現(xiàn)的入口：

class HRegionServer {
    protected RSRpcServices createRpcServices() throws IOException {
        return new RSRpcServices(this);
    }
}

class HMaseter implements HRegionServer {
    @Override
    protected RSRpcServices createRpcServices() throws IOException {
        return new MasterRpcServices(this);
    }
}

RSRpcServices#Constructor

MasterRpcServices 的構(gòu)造方法調(diào)用父類 RSRpcServices 的構(gòu)造方法：

public MasterRpcServices(HMaster m) throws IOException {
    super(m);
    // set HMaster
    master = m;
}
  
public RSRpcServices(HRegionServer rs) throws IOException {
    // set HRegionServer
    regionServer = rs;

    // 初始化 RpcSchedulerFactory
    // 反射 hbase.region.server.rpc.scheduler.factory.class 指定的類
    // 默認(rèn)使用 SimpleRpcSchedulerFactory
    RpcSchedulerFactory rpcSchedulerFactory = ...

    // ...
    
    // 優(yōu)先級函數(shù)（調(diào)度請求時(shí)使用）
    priority = createPriority();
    
    // 設(shè)置訪問時(shí)的重試次數(shù)
    ConnectionUtils.setServerSideHConnectionRetriesConfig(rs.conf, name, LOG);
    
    // 初始化 RpcServer
    try {
        rpcServer = new RpcServer(rs, name, getServices(),
            bindAddress,
            rs.conf,
            rpcSchedulerFactory.create(rs.conf, this, rs));
    } catch (BindException be) {
        // throw Execption
    }

    // 初始化配置信息
    // hbase.client.scanner.timeout.period  # Client 端 Scan Timeout 
    // hbase.server.scanner.max.result.size Scan    # Scan 獲取的最大條目數(shù) 
    // hbase.rpc.timeout    # RPC 處理請求 Timeout 時(shí)間
    // hbase.region.server.rpc.minimum.scan.time.limit.delta    #
    scannerLeaseTimeoutPeriod = rs.conf.getInt(...);
    maxScannerResultSize = rs.conf.getLong(...);
    rpcTimeout = rs.conf.getInt(...);
    minimumScanTimeLimitDelta = rs.conf.getLong(...);

}

RpcServer#Constructor

RpcServer 實(shí)現(xiàn)了 RpcServerInterface 接口矫钓，構(gòu)造函數(shù)：

public RpcServer(final Server server, final String name,
      final List<BlockingServiceAndInterface> services,
      final InetSocketAddress bindAddress, Configuration conf,
      RpcScheduler scheduler)
      throws IOException {
    
    // 初始化屬性 ...

    // 設(shè)置 Listener
    listener = new Listener(name);

    // 設(shè)置 Responder
    responder = new Responder();

    // 設(shè)置 Scheduler（調(diào)用方通過 RpcSchedulerFactory 創(chuàng)建）
    this.scheduler = scheduler;
    this.scheduler.init(new RpcSchedulerContext(this));
}

主要處理角色

Rpc Server 監(jiān)控、讀取舍杜、請求基于 Reactor 模式新娜，主要流程如下圖：

image

Listener

Listener 負(fù)責(zé)監(jiān)聽請求，對于獲取到的請求既绩，交由 Reader 負(fù)責(zé)讀雀帕洹：

private class Listener extends Thread {

    private ServerSocketChannel acceptChannel = null; 
    private Selector selector = null; 
    private Reader[] readers = null;
    private ExecutorService readPool;

    public Listener(final String name) throws IOException {
        // ...

        // 創(chuàng)建非阻塞的 ServerSocketChannel
        acceptChannel = ServerSocketChannel.open();
        acceptChannel.configureBlocking(false);
        
        // 綁定 Socket 到 RpcServer#bingAddress
        bind(acceptChannel.socket(), bindAddress, backlogLength);
        
        // 創(chuàng)建 selector
        selector = Selector.open();

        // 初始化 Reader ThreadPool
        readers = new Reader[readThreads];
        readPool = Executors.newFixedThreadPool(readThreads,
            new ThreadFactoryBuilder()
                .setNameFormat(...)
                .setDaemon(true)
                .build());
        
        for (int i = 0; i < readThreads; ++i) {
            Reader reader = new Reader();
            readers[i] = reader;
            readPool.execute(reader);
        }

        // 注冊 selector
        acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
    }
}

Reader

處理請求的邏輯在 Reader 中，生成 Call 對象交由 RpcSchedule 進(jìn)行分發(fā)

private class Reader implements Runnable {
    private final Selector readSelector;

    @Override
    public void run() {
        try {
            doRunLoop();
        } finally {
            // close readSelector
        }
    }

    private synchronized void doRunLoop() {
        while (running) {
            try {
                // 線程阻塞饲握，知道有請求到來
                readSelector.select();
                
                Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    iter.remove();
                    if (key.isValid()) {
                        if (key.isReadable()) {
                            // 請求有效私杜，進(jìn)行處理
                            doRead(key);
                        }
                    }
                }
            } catch (Exception e) {
                // ...
            }
        }
    }
}

doRead() 方法在 Listener 中，由 Connection 對象處理救欧，生成 Call衰粹，并包裝為 CallRunner 交給 Scheduler

class Listener {
    void doRead(SelectionKey key) throws InterruptedException {
        Connection c = (Connection) key.attachment();
    
        try {
            count = c.readAndProcess();
        } catch (Exception e) {
            // ...
        }
    }
}

class Connection {
    protected void processRequest(byte[] buf) throws IOException, InterruptedException {
        // ...
        
        // Dispatches an RPC request asynchronously
        if (!scheduler.dispatch(new CallRunner(RpcServer.this, call))) {
            
            // ...
            responder.doRespond(call);
        }
    }
}

Scheduler

Scheduler 是一個(gè)生產(chǎn)者消費(fèi)者模型，內(nèi)部有一個(gè)隊(duì)列緩存請求笆怠，另外有一些線程負(fù)責(zé)從隊(duì)列中拉取消息進(jìn)行分發(fā)

Scheduler 默認(rèn)實(shí)現(xiàn)為 SimpleRpcScheduler（HBase 提供的另一種實(shí)現(xiàn)為 FifoRpcScheduler）铝耻，包含三個(gè) RpcExecutor（callExecutor、priorityExecutor蹬刷、replicationExecutor）

public class SimpleRpcScheduler extends RpcScheduler {

    /**
     * callExecutor = RWQueueRpcExecutor or BalancedQueueRpcExecutor
     * priorityExecutor = BalancedQueueRpcExecutor or NULL
     * replicationExecutor = BalancedQueueRpcExecutor or NULL
     **/
    private final RpcExecutor callExecutor;
    private final RpcExecutor priorityExecutor;
    private final RpcExecutor replicationExecutor;
    
    // 分發(fā)請求
    @Override
    public boolean dispatch(CallRunner callTask) throws InterruptedException {
        
        // 選擇不同的 Executor 處理
        // 大部分基于的請求都是通過 callExecutor 來執(zhí)行
        if (priorityExecutor != null && level > highPriorityLevel) {
            return priorityExecutor.dispatch(callTask);
        } else if (replicationExecutor != null && level == HConstants.REPLICATION_QOS) {
            return replicationExecutor.dispatch(callTask);
        } else {
            return callExecutor.dispatch(callTask);
        }
    }
}

RpcExecutor

阻塞隊(duì)列

RpcExecutor 的實(shí)現(xiàn)類 RWQueueRpcExecutor 使用阻塞隊(duì)列緩存消息（BalancedQueueRpcExecutor 實(shí)現(xiàn)類似）：

class RWQueueRpcExecutor extends RpcExecutor {
    private final List<BlockingQueue<CallRunner>> queues;
    
    @Override
    public boolean dispatch(final CallRunner callTask) throws InterruptedException {
        // 進(jìn)入隊(duì)列
        return queues.get(queueIndex).offer(callTask);
    }
}

消費(fèi)線程

RpcExecutor 處理的具體邏輯在 consumerLoop() 方法中瓢捉，從阻塞隊(duì)列中取出 CallRunner 對象，并執(zhí)行：

public abstract class RpcExecutor {

    // 啟動多線程處理
    protected void startHandlers(final String nameSuffix, final int numHandlers,
        final List<BlockingQueue<CallRunner>> callQueues,
        final int qindex, final int qsize, final int port) {
        
        for (int i = 0; i < numHandlers; i++) {
            final int index = qindex + (i % qsize);
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    consumerLoop(callQueues.get(index));
                }
            });
            t.start();
        }
    }

    // 核心處理邏輯（省略部分代碼）
    protected void consumerLoop(final BlockingQueue<CallRunner> myQueue) {
        try {
            while (running) {
                try {
                    // 如果 BlockingQueue 中沒有數(shù)據(jù)办成，會在此阻塞
                    CallRunner task = myQueue.take();
                    try {
                        // 執(zhí)行請求
                        task.run();
                    } catch (Throwable e) {
                        // throw or abort
                    } 
                } catch (InterruptedException e) {
                    interrupted = true;
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

CallRunner & Call

image

CallRunner 和 Call 關(guān)鍵代碼如下：

class CallRunner {
    private Call call;

    public void run() {
        // ...        
        
        Pair<Message, CellScanner> resultPair = null;
        try {
            // make the call
            resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner,
                call.timestamp, this.status);
        } catch (Throwable e) {
            // ...
        } 
        
        // Call to respond
        call.sendResponseIfReady();
    }
}

class Call {
    public synchronized void sendResponseIfReady() throws IOException {
        // 結(jié)果返回給 Client
        this.responder.doRespond(this);
    }
}

調(diào)用執(zhí)行方法在 RpcServer#call() 方法中：

class RpcServer {

    @Override
    public Pair<Message, CellScanner> call(BlockingService service, MethodDescriptor md,
        Message param, CellScanner cellScanner, long receiveTime, MonitoredRPCHandler status)
        throws IOException {
    
    try {
        
        // call method
        // com.google.protobuf#BlockingService
        Message result = service.callBlockingMethod(md, controller, param);
      
        if (tooSlow || tooLarge) {
            // logging
        }
        return new Pair<Message, CellScanner>(result, controller.cellScanner());
    } catch (Throwable e) {
        // ...
    }
  }
}

Responder

Resonder 負(fù)責(zé)發(fā)送 RPC 請求結(jié)果給 Client泡态，Scheduler 調(diào)度請求后，執(zhí)行結(jié)果通過 doRespond() 加入到返回結(jié)果的相應(yīng)隊(duì)列里面

class Responder extends Thread {

    void doRespond(Call call) throws IOException {
        boolean added = false;
        
        // 如果已經(jīng)有一個(gè)正在進(jìn)行的寫入迂卢，不會等待某弦。
        // 這允許立即釋放處理程序以執(zhí)行其他任務(wù)桐汤。
        if (call.connection.responseQueue.isEmpty() && 
            call.connection.responseWriteLock.tryLock()) {
            
            try {
                if (call.connection.responseQueue.isEmpty()) {
                    
                    // 這里如果完成寫操作，直接返回
                    if (processResponse(call)) {
                        return; // we're done.
                    }
                    
                    call.connection.responseQueue.addFirst(call);
                    added = true;
                }
            } finally {
                call.connection.responseWriteLock.unlock();
            }
        }

        if (!added) {
            call.connection.responseQueue.addLast(call);
        }
        
        // Add a connection to the list that want to write,
        call.responder.registerForWrite(call.connection);
    }
}

如果在 doRespond() 中沒有完成寫操作刀崖，通過將 Call 對象的 connection 注冊到 selector惊科，由 Responder 中的線程進(jìn)行后續(xù)的操作。

protected class Responder extends Thread {
    private final Selector writeSelector;

    public void registerForWrite(Connection c) {
        if (writingCons.add(c)) {
            writeSelector.wakeup();
        }
    }
    
    private void doRunLoop() {
        while (running) {
            try {
                // 獲取要寫入的連接列表亮钦，并在 selector 中注冊
                registerWrites();
                
                // ...

                Set<SelectionKey> keys = writeSelector.selectedKeys();
                Iterator<SelectionKey> iter = keys.iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    iter.remove();
                    if (key.isValid() && key.isWritable()) {
                        // 異步寫
                        doAsyncWrite(key);
                    }
                }
            } catch (OutOfMemoryError e) {
                // return or sleep  
            }
        }
    }

    private void doAsyncWrite(SelectionKey key) throws IOException {
        Connection connection = (Connection) key.attachment();
        if (processAllResponses(connection)) {
            // ...
        }
    }

    private boolean processAllResponses(final Connection connection) throws IOException {

        // Only one writer on the channel for a connection at a time.
        connection.responseWriteLock.lock();
        try {
            for (int i = 0; i < 20; i++) {
                Call call = connection.responseQueue.pollFirst();
            
                if (!processResponse(call)) {
                    connection.responseQueue.addFirst(call);
                    return false;
                }
            }
        } finally {
            connection.responseWriteLock.unlock();
        }

        return connection.responseQueue.isEmpty();
    }
}