Long Polling長輪詢實(shí)現(xiàn)進(jìn)階

Long Polling長輪詢實(shí)現(xiàn)進(jìn)階

簡書 滌生。
轉(zhuǎn)載請注明原創(chuàng)出處，謝謝！
如果讀完覺得有收獲的話，歡迎點(diǎn)贊加關(guān)注。

介紹

由于Long Polling長輪詢詳解 這篇文章中的code實(shí)現(xiàn)較為簡單，尤其是服務(wù)端處理較為粗暴，有一些同學(xué)反饋希望服務(wù)端處理阻塞這塊內(nèi)容進(jìn)行更深入討論等等，所以這里專門補(bǔ)一篇實(shí)現(xiàn)進(jìn)階，讓大家對長輪詢有更加深刻的理解。

疑問

對上篇文章，同學(xué)反饋有兩個(gè)疑問。

  • 服務(wù)端實(shí)現(xiàn)使用的是同步servlet，性能比較差，能支撐的連接數(shù)比較少？
    同步servlet來hold請求，確實(shí)會(huì)導(dǎo)致后續(xù)請求得不到及時(shí)處理，servlet3.0開始支持異步處理，可以更高效的處理請求。

  • 服務(wù)端如何去hold住請求，sleep好嗎？
    同步servlet hold住請求的處理邏輯必須在servlet的doGet方法中，一般先fetch數(shù)據(jù)，準(zhǔn)備好了，就返回，沒準(zhǔn)備好，就sleep片刻，再來重復(fù)。
    異步servlet hold住請求比較簡單，只要開啟異步，執(zhí)行完doGet方法后，不會(huì)自動(dòng)返回此次請求，需要等到請求的context被complete，這樣很巧妙的請求就自動(dòng)hold住了。

實(shí)現(xiàn)

  • 客戶端實(shí)現(xiàn)
    客戶端實(shí)現(xiàn)基本和上篇差不多，沒什么改變。
package com.andy.example.longpolling.client;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by andy on 17/7/8.
 */
public class AbstractBootstrap {
    
    //同步URL
    protected static final String URL = "http://localhost:8080/long-polling";
    //異步URL
    protected static final String ASYNC_URL = "http://localhost:8080/long-polling-async";

    private final AtomicLong sequence = new AtomicLong();

    protected void poll() {
        //循環(huán)執(zhí)行，保證每次longpolling結(jié)束，再次發(fā)起longpolling
        while (!Thread.interrupted()) {
            doPoll();
        }
    }

    protected void doPoll() {
        System.out.println("第" + (sequence.incrementAndGet()) + "次 longpolling");

        long startMillis = System.currentTimeMillis();

        HttpURLConnection connection = null;
        try {
            URL getUrl = new URL(URL);
            connection = (HttpURLConnection) getUrl.openConnection();

            //50s作為長輪詢超時(shí)時(shí)間
            connection.setReadTimeout(50000);
            connection.setConnectTimeout(3000);
            connection.setRequestMethod("GET");
            connection.setUseCaches(false);
            connection.setDoOutput(true);
            connection.setDoInput(true);
            connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
            connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");
            connection.connect();

            if (200 == connection.getResponseCode()) {
                BufferedReader reader = null;
                try {
                    reader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
                    StringBuilder result = new StringBuilder(256);
                    String line = null;
                    while ((line = reader.readLine()) != null) {
                        result.append(line);
                    }

                    System.out.println("結(jié)果 " + result);

                } finally {
                    if (reader != null) {
                        reader.close();
                    }
                }
            }
        } catch (IOException e) {
            System.out.println("request failed");
        } finally {
            long elapsed = (System.currentTimeMillis() - startMillis) / 1000;
            System.out.println("connection close" + "     " + "elapse " + elapsed + "s");
            if (connection != null) {
                connection.disconnect();
            }
            System.out.println();
        }
    }

}



package com.andy.example.longpolling.client;

import java.io.IOException;

/**
 * Created by andy on 17/7/6.
 */
public class ClientBootstrap extends AbstractBootstrap {


    public static void main(String[] args) throws IOException {
        ClientBootstrap bootstrap = new ClientBootstrap();
        //發(fā)起longpolling 
        bootstrap.poll();

        System.in.read();
    }

}
  • 服務(wù)端實(shí)現(xiàn)

長輪詢服務(wù)端同步servlet處理

服務(wù)端同步servlet和上篇差不多，沒什么改動(dòng)，增加了相關(guān)注釋。

package com.andy.example.longpolling.server;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by andy on 17/7/6.
 */
@WebServlet(urlPatterns = "/long-polling")
public class LongPollingServlet extends HttpServlet {

    private final Random random = new Random();

    private final AtomicLong sequence = new AtomicLong();

    private final AtomicLong value = new AtomicLong();


    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response)
            throws ServletException, IOException {
        System.out.println();
        final long currentSequence = sequence.incrementAndGet();
        System.out.println("第" + (currentSequence) + "次 longpolling");

        //由于客戶端設(shè)置的超時(shí)時(shí)間是50s，
        //為了更好的展示長輪詢，這邊random 100，模擬服務(wù)端hold住大于50和小于50的情況。
        //再具體場景中，這塊在具體實(shí)現(xiàn)上，
        //對于同步servlet，首先這里必須阻塞，因?yàn)橐坏ヾoGet方法走完，容器就認(rèn)為可以結(jié)束這次請求，返回結(jié)果給客戶端。
        //所以一般實(shí)現(xiàn)如下：
        // while(結(jié)束){ //結(jié)束條件，超時(shí)或者拿到數(shù)據(jù)
        //    data = fetchData();
        //    if(data == null){
        //       sleep();
        //    }
        // }
    
        int sleepSecends = random.nextInt(100);

        System.out.println(currentSequence + " wait " + sleepSecends + " second");

        try {
            TimeUnit.SECONDS.sleep(sleepSecends);
        } catch (InterruptedException e) {

        }

        PrintWriter out = response.getWriter();
        long result = value.getAndIncrement();
        out.write(Long.toString(result));
        out.flush();
    }

}

長輪詢服務(wù)端異步servlet處理

由于同步servlet，性能較差，所有的請求操作必須在doGet方法中完成，包括等待數(shù)據(jù)，占用了容器的處理線程，會(huì)導(dǎo)致后續(xù)的請求阻塞，來不及處理。servlet 3.0支持異步處理，使用異步處理doGet方法執(zhí)行完成后，結(jié)果也不會(huì)返回到客戶端，會(huì)等到請求的context被complete才會(huì)寫回客戶端，這樣一來，容器的處理線程不會(huì)受阻，請求結(jié)果可由另外的業(yè)務(wù)線程進(jìn)行寫回，也就輕松實(shí)現(xiàn)了hold操作。

package com.andy.example.longpolling.server;

import javax.servlet.*;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by andy on 17/7/7.
 */

/**
 * 開啟異步servlet，asyncSupported = true
 */
@WebServlet(urlPatterns = "/long-polling-async", asyncSupported = true)
public class LongPollingAsyncServlet extends HttpServlet {

    private Random random = new Random();

    private final AtomicLong sequence = new AtomicLong();

    private final AtomicLong value = new AtomicLong();

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(100, 200, 50000L,
            TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(100));

    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response)
            throws ServletException, IOException {
        System.out.println();
        final long currentSequence = sequence.incrementAndGet();
        System.out.println("第" + (currentSequence) + "次 longpolling async");
        //設(shè)置request異步處理
        AsyncContext asyncContext = request.startAsync();

        //異步處理超時(shí)時(shí)間，這里需要注意，jetty容器默認(rèn)的這個(gè)值設(shè)置的是30s，
        //如果超時(shí)，異步處理沒有完成（通過是否asyncContext.complete()來進(jìn)行判斷），將會(huì)重試（會(huì)再次調(diào)用doGet方法）。
        //這里由于客戶端long polling設(shè)置的是50s，所以這里如果小于50，會(huì)導(dǎo)致重試。
        asyncContext.setTimeout(51000);
        asyncContext.addListener(new AsyncListener() {
            @Override
            public void onComplete(AsyncEvent event) throws IOException {

            }

            //超時(shí)處理，注意asyncContext.complete();，表示請求處理完成
            @Override
            public void onTimeout(AsyncEvent event) throws IOException {
                AsyncContext asyncContext = event.getAsyncContext();
                asyncContext.complete();
            }

            @Override
            public void onError(AsyncEvent event) throws IOException {

            }

            @Override
            public void onStartAsync(AsyncEvent event) throws IOException {

            }
        });
        
        //提交線程池異步寫會(huì)結(jié)果
        //具體場景中可以有具體的策略進(jìn)行操作
        executor.submit(new HandlePollingTask(currentSequence, asyncContext));

    }

    class HandlePollingTask implements Runnable {

        private AsyncContext asyncContext;

        private long sequense;

        public HandlePollingTask(long sequense, AsyncContext asyncContext) {
            this.sequense = sequense;
            this.asyncContext = asyncContext;
        }

        @Override
        public void run() {
            try {
                //通過asyncContext拿到response
                PrintWriter out = asyncContext.getResponse().getWriter();
                int sleepSecends = random.nextInt(100);


                System.out.println(sequense + " wait " + sleepSecends + " second");

                try {
                    TimeUnit.SECONDS.sleep(sleepSecends);
                } catch (InterruptedException e) {

                }

                long result = value.getAndIncrement();

                out.write(Long.toString(result));

            } catch (Exception e) {
                System.out.println(sequense + "handle polling failed");
            } finally {
                //數(shù)據(jù)寫回客戶端
                asyncContext.complete();
            }
        }
    }

}

結(jié)果

  • 同步servlet實(shí)現(xiàn)結(jié)果
同步servlet客戶端結(jié)果
同步servlet服務(wù)端結(jié)果
  • 異步servlet實(shí)現(xiàn)結(jié)果
異步servlet客戶端結(jié)果
異步servlet服務(wù)端結(jié)果

個(gè)人微信公共號(hào)，感興趣的關(guān)注下，獲取更多技術(shù)文章

滌生-微信公共號(hào)
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • Spring Cloud為開發(fā)人員提供了快速構(gòu)建分布式系統(tǒng)中一些常見模式的工具（例如配置管理，服務(wù)發(fā)現(xiàn)，斷路器，智...
    卡卡羅2017閱讀 136,506評論 19贊 139
  • 基于HTTP的長連接,是一種通過長輪詢方式實(shí)現(xiàn)"服務(wù)器推"的技術(shù),它彌補(bǔ)了HTTP簡單的請求應(yīng)答模式的不足,極大地...
    永遠(yuǎn)的冷冽閱讀 1,203評論 0贊 6
  • 從三月份找實(shí)習(xí)到現(xiàn)在，面了一些公司，掛了不少，但最終還是拿到小米、百度、阿里、京東、新浪、CVTE、樂視家的研發(fā)崗...
    時(shí)芥藍(lán)閱讀 42,759評論 11贊 349
  • 一. Java基礎(chǔ)部分.................................................
    wy_sure閱讀 3,995評論 0贊 11
  • 1. Java基礎(chǔ)部分 基礎(chǔ)部分的順序：基本語法，類相關(guān)的語法，內(nèi)部類的語法，繼承相關(guān)的語法，異常的語法，線程的語...
    子非魚_t_閱讀 34,626評論 18贊 399

友情鏈接更多精彩內(nèi)容