Wie implementiert man einfache lange Abfragen in Java?

Analysieren Sie die Implementierungsmethode für lange Abfragen.

Heutzutage verwenden alle wichtigen Middlewares die Dateninteraktionsmethode für lange Abfragen. Die beliebtesten sind derzeit das Nacos-Konfigurationscenter und der RocketMQ-Pull-Modus. Nachrichten usw. sind alle in der Langabfragemethode implementiert. Wie kann der Server beispielsweise im Nacos-Konfigurationszentrum Konfigurationsänderungen erkennen und sie in Echtzeit an den Client weiterleiten?

Long Polling und Short Polling

#

Kurzabfrage

ist auch ein Pull-Modus. Dies bedeutet, dass der Client unabhängig davon, ob die Serverdaten aktualisiert wurden oder nicht, alle festgelegten Zeiträume zum Abrufen von Daten anfordert. Möglicherweise werden aktualisierte Daten zurückgegeben, oder es werden keine Daten zurückgegeben. Wenn das Konfigurationscenter diese Methode verwendet, treten folgende Probleme auf: Da sich die Konfigurationsdaten nicht häufig ändern, wird der Server zwangsläufig stark belastet, wenn Sie weiterhin Anfragen senden. Dies führt beispielsweise auch zu einer Verzögerung beim Pushen von Daten: Wenn die Konfiguration in der 11. Sekunde aktualisiert wird, wird der Push um 9 Sekunden verzögert und auf die nächste Anfrage gewartet 🎜#

Neutralisierung

zwischen Push-Latenz und Serverdruck nicht möglich. Wenn Sie das Abfrageintervall verringern, nimmt die Verzögerung ab und der Druck zu. Wenn Sie das Abfrageintervall erhöhen, verringert sich der Druck und die Verzögerung nimmt zu.

Lange AbfragenUm das Problem der kurzen Abfragen zu lösen, initiiert der Client eine lange Abfrage. Wenn sich die Daten auf dem Server nicht ändern, wird die Anfrage zurückgehalten Der Dienst ändert die Daten am Ende oder wartet eine bestimmte Zeit, bevor er zurückkehrt. Nach der Rückkehr initiiert der Client die nächste lange Abfrageanforderung zum Abhören.

Die Vorteile dieses Designs:

Im Vergleich zur geringen Latenz initiiert der Client eine lange Abfrage und der Server erkennt, dass sich die Daten geändert haben . kann sofort eine Antwort an den Client zurücksenden.

• Der Druck auf den Server wird reduziert und der Client initiiert eine lange Abfrage. Wenn sich die Daten nicht ändern, hält der Server die Anfrage des Clients zurück Die Zeit ist im Allgemeinen auf 30 oder 60 Sekunden eingestellt, und der Server hält die Anforderung, ohne zu viele Serverressourcen zu verbrauchen.

• Die folgenden Bilder dienen zur Veranschaulichung des Vorgangs:

#🎜 🎜#Zuerst initiiert der Client eine lange Abfrageanfrage. Wenn der Server die Anfrage des Clients innerhalb der vom Server festgelegten Zeitspanne unterbricht, antwortet der Server dem Client Wenn sich keine Änderung ergibt, wird der Client weiterhin Anfragen senden.

• Wenn sich die Dienstdaten innerhalb von 30 Sekunden ändern, überträgt der Server die geänderten Daten an den Client. #? Implementieren Sie es mit Code:

• Zuerst sendet der Client eine HTTP-Anfrage an den Server; der Server startet einen asynchronen Thread und dieser bleibt hängen, wenn keine Daten vorhanden sind Die aktuelle Anforderung (ein Tomcat hat 200 Threads, lange Abfragen sollten die Geschäftsthreads von Tomcat nicht blockieren, daher verwendet das Konfigurationscenter bei der Implementierung langer Abfragen häufig asynchrone Antworten, was für die Implementierung von asynchronem HTTP-Common bequemer ist. Die Methode ist #🎜🎜 #AsyncContext-Mechanismus, der von Servlet3.0 bereitgestellt wird. Er gibt eine unveränderte Identität an den Client zurück. Reagieren Sie beispielsweise auf den Statuscode 304; ;

  #🎜🎜 #

Konfigurationscenter-Long-Polling-Implementierung

Der folgende Code wird zum Implementieren von Long-Polling verwendet:

Client-Implementierung

 @Slf4j
 public class ConfigClientWorker {
 
     private final CloseableHttpClient httpClient;
 
     private final ScheduledExecutorService executorService;
 
     public ConfigClientWorker(String url, String dataId) {
         this.executorService = Executors.newSingleThreadScheduledExecutor(runnable -> {
             Thread thread = new Thread(runnable);
             thread.setName("client.worker.executor-%d");
             thread.setDaemon(true);
             return thread;
         });
 
         // ① httpClient 客户端超时时间要大于长轮询约定的超时时间
         RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(40000).build();
         this.httpClient = HttpClientBuilder.create().setDefaultRequestConfig(requestConfig).build();
 
         executorService.execute(new LongPollingRunnable(url, dataId));
     }
 
     class LongPollingRunnable implements Runnable {
 
         private final String url;
         private final String dataId;
 
         public LongPollingRunnable(String url, String dataId) {
             this.url = url;
             this.dataId = dataId;
         }
 
         @SneakyThrows
         @Override
         public void run() {
             String endpoint = url + "?dataId=" + dataId;
             log.info("endpoint: {}", endpoint);
             HttpGet request = new HttpGet(endpoint);
             CloseableHttpResponse response = httpClient.execute(request);
             switch (response.getStatusLine().getStatusCode()) {
                 case 200: {
                     BufferedReader rd = new BufferedReader(new InputStreamReader(response.getEntity()
                             .getContent()));
                     StringBuilder result = new StringBuilder();
                     String line;
                     while ((line = rd.readLine()) != null) {
                         result.append(line);
                     }
                     response.close();
                     String configInfo = result.toString();
                     log.info("dataId: [{}] changed, receive configInfo: {}", dataId, configInfo);
                     break;
                 }
                 // ② 304 响应码标记配置未变更
                 case 304: {
                     log.info("longPolling dataId: [{}] once finished, configInfo is unchanged, longPolling again", dataId);
                     break;
                 }
                 default: {
                     throw new RuntimeException("unExcepted HTTP status code");
                 }
             }
             executorService.execute(this);
         }
     }
 
     public static void main(String[] args) throws IOException {
 
         new ConfigClientWorker("http://127.0.0.1:8080/listener", "user");
         System.in.read();
     }
 }

#🎜🎜 #

httpClient Das Client-Timeout muss größer sein als das durch lange Abfragen vereinbarte Timeout, sonst kommt es beim Client selbst zu einem Timeout, bevor der Server zurückkehrt.

• 304 Antwortcode-Tag-Konfiguration hat sich nicht geändert;

http://127.0.0.1:8080 / Listener ist die Serveradresse; Der asynchrone Thread request.startAsync(request, Response); belegt garantiert nicht den Tomcat-Thread. Zu diesem Zeitpunkt wird der Tomcat-Thread veröffentlicht. Wird mit asyncContext.complete() verwendet.
dataIdContext.put(dataId, asyncTask); ordnet die dataId dem asynchronen Anforderungskontext zu, um die Konfiguration und Freigabe des entsprechenden Kontexts zu erleichtern. 🎜🎜#

• Multimap<string asynctask> dataIdContext</string>Ein Schlüssel kann mehreren Werten entsprechen verstanden werden als Map<string>></string>

timeoutChecker.schedule() Startzeit Der Server schreibt nach 30 Sekunden eine 304-Antwort

@RequestMapping("/publishConfig") ，配置发布的入口。配置变更后，根据 dataId 一次拿出所有的长轮询，为之写入变更的响应。

asyncTask.getAsyncContext().complete();表示这次异步请求结束了。

启动配置监听

先启动 ConfigServer，再启动 ConfigClient。30s之后控制台打印第一次超时之后收到服务端304的状态码

 16:41:14.824 [client.worker.executor-%d] INFO cn.haoxiaoyong.poll.ConfigClientWorker - longPolling dataId: [user] once finished, configInfo is unchanged, longPolling again

请求一下配置发布，请求localhost:8080/publishConfig?dataId=user&configInfo=helloworld

服务端打印日志：

 2022-08-25 16:45:56.663  INFO 90650 --- [nio-8080-exec-2] cn.haoxiaoyong.poll.ConfigServer         : publish configInfo dataId: [user], configInfo: helloworld

Das obige ist der detaillierte Inhalt vonWie implementiert man einfache lange Abfragen in Java?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!