我在Elasticsearch上遇到问题，我不希望对索引项进行分析。但是elasticsearch有一些默认设置，可以在空间上标记它。因此，我的方面查询未返回我想要的结果。

我读到"index" : "not_analyzed"索引类型的属性应该工作。但是问题是我事先不知道我的文档结构。我会在不知道表结构的情况下将随机MySQL数据库索引到elasticsearch。

我如何设置elasticsearch，使其默认情况下会"index" : "not_analyzed"一直使用，除非另有要求。谢谢

PS：如果我可以直接使用任何API，我会使用Java。

injector.getInstance(HttpServerTransport.class).start();//提供HttpServerTransport服务的启动

最终调用Netty4HttpServerTransport中的doStart()方法 

serverBootstrap = new ServerBootstrap();//即Netty的服务端启动方式

serverBootstrap.childHandler(configureServerChannelHandler());//添加服务端接消息的处理类

this.requestHandler = new Netty4HttpRequestHandler(transport, detailedErrorsEnabled, threadContext);

ch.pipeline().addLast("handler", requestHandler);

protected void doStart() {
        boolean success = false;
        try {
            this.serverOpenChannels = new Netty4OpenChannelsHandler(logger);

            serverBootstrap = new ServerBootstrap();
            if (blockingServer) {
                serverBootstrap.group(new OioEventLoopGroup(workerCount, daemonThreadFactory(settings,
                    HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)));
                serverBootstrap.channel(OioServerSocketChannel.class);
            } else {
                serverBootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings,
                    HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)));
                serverBootstrap.channel(NioServerSocketChannel.class);
            }

            serverBootstrap.childHandler(configureServerChannelHandler());

            serverBootstrap.childOption(ChannelOption.TCP_NODELAY, SETTING_HTTP_TCP_NO_DELAY.get(settings));
            serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, SETTING_HTTP_TCP_KEEP_ALIVE.get(settings));

            final ByteSizeValue tcpSendBufferSize = SETTING_HTTP_TCP_SEND_BUFFER_SIZE.get(settings);
            if (tcpSendBufferSize.getBytes() > 0) {
                serverBootstrap.childOption(ChannelOption.SO_SNDBUF, Math.toIntExact(tcpSendBufferSize.getBytes()));
            }

            final ByteSizeValue tcpReceiveBufferSize = SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE.get(settings);
            if (tcpReceiveBufferSize.getBytes() > 0) {
                serverBootstrap.childOption(ChannelOption.SO_RCVBUF, Math.toIntExact(tcpReceiveBufferSize.getBytes()));
            }

            serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator);
            serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator);

            final boolean reuseAddress = SETTING_HTTP_TCP_REUSE_ADDRESS.get(settings);
            serverBootstrap.option(ChannelOption.SO_REUSEADDR, reuseAddress);
            serverBootstrap.childOption(ChannelOption.SO_REUSEADDR, reuseAddress);

            this.boundAddress = createBoundHttpAddress();
            if (logger.isInfoEnabled()) {
                logger.info("{}", boundAddress);
            }
            success = true;
        } finally {
            if (success == false) {
                doStop(); // otherwise we leak threads since we never moved to started
            }
        }
    }

View Code

TransportService transportService = injector.getInstance(TransportService.class);//提供Transport服务的启动

transportService.start();

最终调用Netty4Transport中的doStart()方法

bootstrap = createBootstrap();//客户端启动

bootstrap.handler(getClientChannelInitializer());//添加客户端消息处理类

ch.pipeline().addLast("dispatcher", new Netty4MessageChannelHandler(Netty4Transport.this, ".client"));

createServerBootstrap(entry.getKey(), settings);//服务端启动

serverBootstrap.childHandler(getServerChannelInitializer(name, settings));//添加服务端消息处理类

ch.pipeline().addLast("dispatcher", new Netty4MessageChannelHandler(Netty4Transport.this, name));

    @Override
    protected void doStart() {
        boolean success = false;
        try {
            bootstrap = createBootstrap();
            if (NetworkService.NETWORK_SERVER.get(settings)) {
                final Netty4OpenChannelsHandler openChannels = new Netty4OpenChannelsHandler(logger);
                this.serverOpenChannels = openChannels;
                // loop through all profiles and start them up, special handling for default one
                for (Map.Entry<String, Settings> entry : buildProfileSettings().entrySet()) {
                    // merge fallback settings with default settings with profile settings so we have complete settings with default values
                    final Settings settings = Settings.builder()
                        .put(createFallbackSettings())
                        .put(entry.getValue()).build();
                    createServerBootstrap(entry.getKey(), settings);
                    bindServer(entry.getKey(), settings);
                }
            }
            super.doStart();
            success = true;
        } finally {
            if (success == false) {
                doStop();
            }
        }
    }

View Code

    @Override
    protected <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void doExecute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) {
        proxy.execute(action, request, listener);
    }

final TransportActionNodeProxy<Request, Response> proxy = proxies.get(action);//根据action类型不同，生产执行不同的execute方法
nodesService.execute((n, l) -> proxy.execute(n, request, l), listener);

transportService.sendRequest(node, action.name(), request, transportOptions,
    new ActionListenerResponseHandler<>(listener, action::newResponse));
}

Transport.Connection connection = getConnection(node);//节点连接
sendRequest(connection, action, request, options, handler);//发送请求

connection.sendRequest(requestId, action, request, options);

通过channel发送请求
Channel channel = channel(options.type());
sendRequestToChannel(this.node, channel, requestId, action, request, options, getVersion(), (byte)0);

最终执行Netty4Transport的sendMessage方法将请求发送至channel

@Override
protected void sendMessage(Channel channel, BytesReference reference, Runnable sendListener) {
    final ChannelFuture future = channel.writeAndFlush(Netty4Utils.toByteBuf(reference));
    future.addListener(f -> sendListener.run());
}

transport.messageReceived(reference, ctx.channel(), profileName, remoteAddress, remainingMessageSize);

final RequestHandlerRegistry reg = transportServiceAdapter.getRequestHandler(action);//根据action类型，生产具体的TransportRequest子类

final TransportRequest request = reg.newRequest();

threadPool.executor(reg.getExecutor()).execute(new RequestHandler(reg, request, transportChannel));

        @Override
        protected void doRun() throws Exception {
            reg.processMessageReceived(request, transportChannel);//
        }

handler.messageReceived(request, channel);//TransportRequestHandler中的messageReceived方法

最终调用TransportHandler中的messageReceived方法
        public final void messageReceived(final Request request, final TransportChannel channel, Task task) throws Exception {
            // We already got the task created on the network layer - no need to create it again on the transport layer
            execute(task, request, new ActionListener<Response>() {//执行具体的请求处理
                @Override
                public void onResponse(Response response) {
                    try {
                        channel.sendResponse(response);
                    } catch (Exception e) {
                        onFailure(e);
                    }
                }
.......

public ActionFuture<BulkResponse> bulk(final BulkRequest request) {
        return execute(BulkAction.INSTANCE, request);
    }

final TransportActionNodeProxy<Request, Response> proxy = proxies.get(action);//proxy则根据不同的action，生成不同的TransportActionNodeProxy

nodesService.execute((n, l) -> proxy.execute(n, request, l), listener);//nodesService为TransportClientNodesService实例

public <Response> void execute(NodeListenerCallback<Response> callback, ActionListener<Response> listener) {
        // we first read nodes before checking the closed state; this
        // is because otherwise we could be subject to a race where we
        // read the state as not being closed, and then the client is
        // closed and the nodes list is cleared, and then a
        // NoNodeAvailableException is thrown
        // it is important that the order of first setting the state of
        // closed and then clearing the list of nodes is maintained in
        // the close method
        final List<DiscoveryNode> nodes = this.nodes;
        if (closed) {
            throw new IllegalStateException("transport client is closed");
        }
        ensureNodesAreAvailable(nodes);
        int index = getNodeNumber();
        RetryListener<Response> retryListener = new RetryListener<>(callback, listener, nodes, index, hostFailureListener);
        DiscoveryNode node = retryListener.getNode(0);
        try {
            callback.doWithNode(node, retryListener);
        } catch (Exception e) {
            try {
                //this exception can''t come from the TransportService as it doesn''t throw exception at all
                listener.onFailure(e);
            } finally {
                retryListener.maybeNodeFailed(node, e);
            }
        }
    }

final DiscoveryNode getNode(int i) {
            return nodes.get((index + i) % nodes.size());
        }

public void execute(final DiscoveryNode node, final Request request, final ActionListener<Response> listener) {
        ActionRequestValidationException validationException = request.validate();
        if (validationException != null) {
            listener.onFailure(validationException);
            return;
        }
        transportService.sendRequest(node, action.name(), request, transportOptions,
            new ActionListenerResponseHandler<>(listener, action::newResponse));//根据请求类型调用sendRequest方法
    }

public final <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,
                                                                final TransportRequest request,
                                                                final TransportRequestOptions options,
                                                                TransportResponseHandler<T> handler) {
        try {
            Transport.Connection connection = getConnection(node);//连接节点
            sendRequest(connection, action, request, options, handler);//处理请求
        } catch (NodeNotConnectedException ex) {
            // the caller might not handle this so we invoke the handler
            handler.handleException(ex);
        }
    }

        @Override
        public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
            throws IOException, TransportException {
            if (closed.get()) {
                throw new NodeNotConnectedException(node, "connection already closed");
            }
            Channel channel = channel(options.type());//根据不同的请求类型，添加不同的过滤器
            sendRequestToChannel(this.node, channel, requestId, action, request, options, getVersion(), (byte)0);
        }
    }

private boolean internalSendMessage(Channel targetChannel, BytesReference message, Runnable onRequestSent) throws IOException {
        boolean success;
        try {
            sendMessage(targetChannel, message, onRequestSent);//终极调用Netty4Transport的sendMessage方法
            success = true;
        } catch (IOException ex) {
            // passing exception handling to deal with this and raise disconnect events and decide the right logging level
            onException(targetChannel, ex);
            success = false;
        }
        return success;
    }

    @Override
    protected void sendMessage(Channel channel, BytesReference reference, Runnable sendListener) {
        final ChannelFuture future = channel.writeAndFlush(Netty4Utils.toByteBuf(reference));//向服务端发送消息
        future.addListener(f -> sendListener.run());