Small knowledge, big challenge! This article is participating in the creation activity of “Essential Tips for Programmers”

This article has participated in the “Digitalstar Project” and won a creative gift package to challenge the creative incentive money.

Nacos configures the server – side long polling processing mechanism in the center

The client initiates an HTTP request for a long poll, invoking the interface/Listner server in the ConfigController class of the Nacos-Config module:

The listener method for ConfigController:

/** * compare MD5 */
@PostMapping("/listener")
public void listener(HttpServletRequest request, HttpServletResponse response)
    throws ServletException, IOException {
    request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED".true);
    String probeModify = request.getParameter("Listening-Configs");
    if (StringUtils.isBlank(probeModify)) {
        throw new IllegalArgumentException("invalid probeModify");
    }

    probeModify = URLDecoder.decode(probeModify, Constants.ENCODE);

    Map<String, String> clientMd5Map;
    try {
        clientMd5Map = MD5Util.getClientMd5Map(probeModify);
    } catch (Throwable e) {
        throw new IllegalArgumentException("invalid probeModify");
    }

    // do long-polling
    inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}
Copy the code
  1. Obtain the configurations that the client needs to listen for possible transmission changes and calculate the MD5 value
  2. Inner. DoPollingConfig executive poll request

doPollingConfig

/** * polling interface */
public String doPollingConfig(HttpServletRequest request, HttpServletResponse response,
                              Map<String, String> clientMd5Map, int probeRequestSize)
    throws IOException {

    / / long polling
    if (LongPollingService.isSupportLongPolling(request)) {
        longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
        return HttpServletResponse.SC_OK + "";
    }

    // else compatible with short polling logic
    List<String> changedGroups = MD5Util.compareMd5(request, response, clientMd5Map);

    // Compatible with short polling result
    String oldResult = MD5Util.compareMd5OldResult(changedGroups);
    String newResult = MD5Util.compareMd5ResultString(changedGroups);

    String version = request.getHeader(Constants.CLIENT_VERSION_HEADER);
    if (version == null) {
        version = "2.0.0";
    }
    int versionNum = Protocol.getVersionNumber(version);

    /** * Before version 2.0.4, return values were placed in the header */
    if (versionNum < START_LONGPOLLING_VERSION_NUM) {
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE, oldResult);
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE_NEW, newResult);
    } else {
        request.setAttribute("content", newResult);
    }

    // Disable caching
    response.setHeader("Pragma"."no-cache");
    response.setDateHeader("Expires".0);
    response.setHeader("Cache-Control"."no-cache,no-store");
    response.setStatus(HttpServletResponse.SC_OK);
    return HttpServletResponse.SC_OK + "";
}
Copy the code
  1. Determine whether the current request is a long poll and, if so, call addLongPollingClient

addLongPollingClient

The client’s long polling request is mainly encapsulated as ClientLongPolling and sent to scheduler for execution

public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
                                 int probeRequestSize) {

    String str = req.getHeader(LongPollingService.LONG_POLLING_HEADER);
    String noHangUpFlag = req.getHeader(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER);
    String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
    String tag = req.getHeader("Vipserver-Tag");
    int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
    /** * Return response 500ms earlier to avoid client timeout@qiaoyiDingqy 2013.10.22 modified Add delay time for LoadBalance */
    long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
    if (isFixedPolling()) {
        timeout = Math.max(10000, getFixedPollingInterval());
        // do nothing but set fix polling timeout
    } else {
        long start = System.currentTimeMillis();
        List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
        if (changedGroups.size() > 0) {
            generateResponse(req, rsp, changedGroups);
            LogUtil.clientLog.info(| | | "{} {} {} {} | | {} {} | {}",
                System.currentTimeMillis() - start, "instant", RequestUtil.getRemoteIp(req), "polling",
                clientMd5Map.size(), probeRequestSize, changedGroups.size());
            return;
        } else if(noHangUpFlag ! =null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
            LogUtil.clientLog.info(| | | "{} {} {} {} | | {} {} | {}", System.currentTimeMillis() - start, "nohangup",
                RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
                changedGroups.size());
            return;
        }
    }
    String ip = RequestUtil.getRemoteIp(req);
    // It must be called by an HTTP thread, otherwise the container will send the response immediately after leaving
    final AsyncContext asyncContext = req.startAsync();
    // asyncContext.setTimeout () is invalid, so you can only control it yourself
    asyncContext.setTimeout(0L);

    scheduler.execute(
        new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}
Copy the code
  1. Get the timeout of the client request, subtract 500ms and assign to the timeout variable.
  2. Check isFixedPolling. If isFixedPolling is true, the scheduled task will be executed 30 seconds later. Otherwise, the scheduled task will be executed 29.5 seconds later
  3. MD5 is used to compare the data with that on the server. If changes are sent, the message is returned directly
  4. Scheduler. execute executes the ClientLongPolling thread

ClientLongPolling

ClientLongPolling is a thread with the run method as follows:

@Override
public void run(a) {
    asyncTimeoutFuture = scheduler.schedule(new Runnable() {
        @Override
        public void run(a) {
            try {
                getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                /** * Delete subscription */
                allSubs.remove(ClientLongPolling.this);

                if (isFixedPolling()) {
                    LogUtil.clientLog.info(| | | "{} {} {} {} | | {} {}",
                        (System.currentTimeMillis() - createTime),
                        "fix", RequestUtil.getRemoteIp((HttpServletRequest)asyncContext.getRequest()),
                        "polling",
                        clientMd5Map.size(), probeRequestSize);
                    List<String> changedGroups = MD5Util.compareMd5(
                        (HttpServletRequest)asyncContext.getRequest(),
                        (HttpServletResponse)asyncContext.getResponse(), clientMd5Map);
                    if (changedGroups.size() > 0) {
                        sendResponse(changedGroups);
                    } else {
                        sendResponse(null); }}else {
                    LogUtil.clientLog.info(| | | "{} {} {} {} | | {} {}",
                        (System.currentTimeMillis() - createTime),
                        "timeout", RequestUtil.getRemoteIp((HttpServletRequest)asyncContext.getRequest()),
                        "polling",
                        clientMd5Map.size(), probeRequestSize);
                    sendResponse(null); }}catch (Throwable t) {
                LogUtil.defaultLog.error("long polling error:" + t.getMessage(), t.getCause());
            }

        }

    }, timeoutTime, TimeUnit.MILLISECONDS);

    allSubs.add(this);
}
Copy the code
  1. A scheduled task is started through scheduler.schedule with a delay of 29.5s
  2. The ClientLongPolling instance itself is added to the allSubs queue, which primarily maintains a long-polling subscription.
  3. After the scheduled task is executed, the ClientLongPolling instance itself is removed from the allSubs queue.
  4. MD5 is used to compare whether the groupKeys requested by the client are changed, and the change result is returned to the client through response

The so-called long polling means that the server does not return the request immediately after receiving it, but returns the request result to the client 29.5 seconds later. In this way, the client and the server remain connected within 30 seconds without any data changes.