跟正常的網路通訊相似，es的client跟server是通過netty進行通訊的，client封裝各種request，通過netty傳送給es的server。server解析收到的各類request，dispatch到對應的handler中進行處理。

下面我們看一下索引一條資料設計到的程式碼呼叫。

客戶端當然就是發起請求的了：client.prepareIndex("index","type","id").execute(); 我們假設使用的是TransportClient。過程如下：

prepareIndex會生成一個IndexRequestBuilder，其execute為基類ActionRequestBuilder的呼叫：

  public void execute(ActionListener<Response> listener) {
        doExecute(listener);
    }

IndexRequestBuilder覆蓋doExecute方法：

protected void doExecute(ActionListener<IndexResponse> listener) {
        client.index(request, listener);
    }

呼叫到client的index方法裡邊去了，其中client我們使用的TransportClient（裡邊使用的是InternalTransportClient例項）：

  public ActionFuture<IndexResponse> index(IndexRequest request) {
        return internalClient.index(request);
    }
 

上述呼叫會進入TranportClient的基類AbstractClient中：

public ActionFuture<IndexResponse> index(final IndexRequest request) {
        return execute(IndexAction.INSTANCE, request);
    }

注意這裡的IndexAction已經在Client初始化的時候繫結到了TransportIndexAction

modules.add(new ActionModule(true));

registerAction(IndexAction.INSTANCE, TransportIndexAction.class);
 

這個基類的方法在InternalTransportClient的實現如下：

  public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder, Client>> ActionFuture<Response> execute(final Action<Request, Response, RequestBuilder, Client> action, final Request request) {
        final TransportActionNodeProxy<Request, Response> proxy = actions.get(action);
        return nodesService.execute(new TransportClientNodesService.NodeCallback<ActionFuture<Response>>() {
            @Override
            public ActionFuture<Response> doWithNode(DiscoveryNode node) throws ElasticsearchException {
                return proxy.execute(node, request);
            }
        });
    }

在InternalTransportClient內部維護這樣一個結構：

private final ImmutableMap<Action, TransportActionNodeProxy> actions;

註冊了action跟TransportActionNodeProxy的對應關係。因此，在execute的過程中會根據action取對應的TransportActionNodeProxy例項（註冊了），最終呼叫到execute方法，在該方法中借用transportServive通訊模組把請求發給伺服器：

   transportService.sendRequest(node, action.name(), request, transportOptions, new BaseTransportResponseHandler<Response>()

至此，進入伺服器端的處理了：

nettyTranspot中dispatch如下：

pipeline.addLast("dispatcher", new MessageChannelHandler(NettyTransport.this, logger));

進入MessageChannelHandler的messageReceived，一通解析之後，調入handleRequest中：

TransportRequestHandler handler = transportServiceAdapter.handler(action);

TransportIndexAction對應的handler是OperationTransportHandler？在TransportShardReplicationOperationAction中註冊了：

transportService.registerHandler(actionName, new OperationTransportHandler());

至此，已經找到訊息的處理函式，進入OperationTransportHandler中，執行messageReceived方法：

 public void messageReceived(final Request request, final TransportChannel channel) throws Exception {
            // no need to have a threaded listener since we just send back a response
            request.listenerThreaded(false);
            // if we have a local operation, execute it on a thread since we don't spawn
            request.operationThreaded(true);
            execute(request, new ActionListener<Response>() {
                @Override
                public void onResponse(Response result) {
                    try {
                        channel.sendResponse(result);
                    } catch (Throwable e) {
                        onFailure(e);
                    }
                }

                @Override
                public void onFailure(Throwable e) {
                    try {
                        channel.sendResponse(e);
                    } catch (Throwable e1) {
                        logger.warn("Failed to send response for " + actionName, e1);
                    }
                }
            });
        }

具體執行過程到execute方法中，該方法是基類TransportAction的方法，進一步檢視基類execute方法，可以發現最終又回到了TransportIndexAction的doExecute方法，然後執行InnerExecute進一步執行基類TransportShardReplicationOperationAction的doExecute方法，最終歸宿為AsyncShardOperationAction：

 protected void doExecute(Request request, ActionListener<Response> listener) {
        new AsyncShardOperationAction(request, listener).start();
    }

over。經過一番周折才找到最終服務端的執行函式AsyncShardOperationAction。。。。

接下來的操作就是真正的索引操作了，函式都集中在TransportIndexAction和基類TransportShardReplicationOperationAction中，來回穿梭。。。

AsyncShardOperationAction的doStart方法為入口，讀取叢集狀態，獲取索引請求需要在哪些shard上執行（根據indexname，type，id，routing值），找到primary shard和一致性保障，然後分別在primary shards和replicas上執行操作。其中TransportAction中的shardOperationOnPrimary函式為在primary上執行的操作。

shardOperationOnPrimary中會判斷操作的型別：

 if (request.opType() == IndexRequest.OpType.INDEX) {
            Engine.Index index = indexShard.prepareIndex(sourceToParse, request.version(), request.versionType(), Engine.Operation.Origin.PRIMARY, request.canHaveDuplicates());
            if (index.parsedDoc().mappingsModified()) {
                mappingUpdatedAction.updateMappingOnMaster(request.index(), index.docMapper(), indexService.indexUUID());
            }
            indexShard.index(index);
            version = index.version();
            op = index;
            created = index.created();
        } else {
            Engine.Create create = indexShard.prepareCreate(sourceToParse,
                    request.version(), request.versionType(), Engine.Operation.Origin.PRIMARY, request.canHaveDuplicates(), request.autoGeneratedId());
            if (create.parsedDoc().mappingsModified()) {
                mappingUpdatedAction.updateMappingOnMaster(request.index(), create.docMapper(), indexService.indexUUID());
            }
            indexShard.create(create);
            version = create.version();
            op = create;
            created = true;
        }

可見分為兩種，一種是index操作（id已經存在，就更新doc的值），一種是create操作（建立新的doc）.

InternalIndexShard執行index操作：其中會根據source內容來判定索引資料是否對mapping進行了更改，如果有更改，會updateMappingOnMaster，然後呼叫InternalEngine的index方法執行lucene的Index操作：執行一些version的操作，是否衝突，確定最終的version，然後寫入lucene索引，寫入translog。

主分片操作完成之後，在副本上執行操作。

至此，一條索引資料完成了。。。。。