Jetty : Embedded Server 啟動流程 - 1

everfly 發(fā)布于2019-08-16 12:23 / 2670人閱讀

前言

本文基于 Jetty 8.1.x 版本簡單介紹 Jetty Embedded Server 核心概念，線程模型，啟動流程。以下代碼片段摘自 Jetty 源代碼中的 example-jetty-embedded 模塊的 OneServletContext.java

public class OneServletContext {
    public static void main(String[] args) throws Exception {
        Server server = new Server(8080);

        ServletContextHandler context = new ServletContextHandler(
        ServletContextHandler.SESSIONS);
        context.setContextPath("/");
        server.setHandler(context);

        ...
        
        // Serve some hello world servlets
        context.addServlet(new ServletHolder(new HelloServlet()),"/*");
        context.addServlet(new ServletHolder(
                new HelloServlet("Buongiorno Mondo")),"/it/*");
        context.addServlet(new ServletHolder(
                new HelloServlet("Bonjour le Monde")),"/fr/*");

        server.start();
        server.join();
    }
}

Embedded Jetty 使用起來很方便，少數(shù)幾行代碼就可以實(shí)現(xiàn)一個 http （Servlet）接口

創(chuàng)建 Server 對象，設(shè)置要監(jiān)聽的端口 8080

創(chuàng)建 ServletContextHandler 對象，ServletContextHandler "is-a" Handler（Request Handler），提供 ServletContext

將 ServletContextHandler 關(guān)聯(lián)到 Server(即 server.setHandler(context))

向 ServletContextHandler 中添加 Servlet，以 ServletHolder 的形式

啟動 Server

核心概念 LifeCycle

The lifecycle（生命周期） interface for generic components，聲明 start（啟動），stop（停止），isRunning，isStarted（查詢狀態(tài)）等方法

AbstractLifeCycle

實(shí)現(xiàn) LifeCycle 接口的抽象類，提供生命周期管理默認(rèn)實(shí)現(xiàn):例如通過 _state 屬性保存組件狀態(tài)，提供 start，stop 默認(rèn)實(shí)現(xiàn)

public final void start() throws Exception {
    // 線程安全
    synchronized (_lock) {
        try {
            // 狀態(tài)保護(hù)
            if (_state == _STARTED || _state == _STARTING) {
                return;
            }
            setStarting();
            // 將具體的 start 邏輯推遲到具體的子類實(shí)現(xiàn)
            doStart();
            setStarted();
        } catch (...) {
            ...
        }
    }
}

AggregateLifeCycle

An AggregateLifeCycle is an LifeCycle implementation for a collection of contained beans
AggregateLifeCycle 繼承自 AbstractLifeCycle，管理一組 Bean 的生命周期

public class AggregateLifeCycle extends AbstractLifeCycle implements
    Destroyable, Dumpable {
    private final List _beans = new CopyOnWriteArrayList<>();

    private boolean _started = false;

    @Override
    protected void doStart() throws Exception {
        for (Bean b: beans) {
            if (b._managed && b._bean instanceof LifeCycle) {
                LifeCycle l = (LefeCycle) b._bean;
                if (!l.isRunning()) {
                    // start managed bean
                    l.start();
                }
            }
        }
        _started =true;
        super.doStart();
    }
}

Server

類層次

AbstractLifeCycle
    AggregateLifeCycle
        AbstractHandler
            AbstractHandlerContainer
                HandlerWrapper
                    Server

Server 類是 Jetty 的《門面》類，包含:

Connector，接收客戶端請求

Handler，處理客戶端請求

ThreadPool，任務(wù)調(diào)度線程池

通過 Server 類提供的各種 set 屬性方法可以定制 Connector, ThreadPool，如果沒有特殊指定 Server 類會創(chuàng)建默認(rèn)實(shí)現(xiàn)，比如默認(rèn)的 Connector 為 SelectChannelConnector，默認(rèn)的 ThreadPool 為 QueuedThreadPool

Connector

（主要）類層次：

Connector
    AbstractConnector
        AbstractNIOConnector
            SelectChannelConnector

Connector 接口及其實(shí)現(xiàn)類用于接收客戶端請求，（HTTP/SPDY）協(xié)議解析，最終調(diào)用 Server 中的（Request）Handler 處理請求，SelectChannelConnector 是 Server 默認(rèn)使用的 Connector

public Server(int port) {
    setServer(this);

    Connector connector = new SelectChannelConnector();
    connector.setPort(port);
    setConnectors(new Connector[]{ connector });
}

Handler

Handler 或者叫作 Request Handler，用于處理客戶端請求

public interface Handler extends LifeCycle, Destroyable {
    ...
    void handle(String target, Request baseRequest, HttpServletRequest request,
            HttpServletResponse response);
    ...
}

handle 方法的簽名已經(jīng)很接近 Servlet service 方法，這里的 target 通常是請求的 uri，用于根據(jù) Servlet 配置規(guī)則找到具體的 Servlet 并調(diào)用其 service 方法

線程模型

Jetty Server 使用線程池來處理客戶端請求，ThreadPool 接口定義了線程池基本操作，Server 默認(rèn)使用 QueuedThreadPool 類，如果需要，可以通過 Server.setThreadPool 方法手動設(shè)置線程池
注：Jetty 9.x 線程模型比較復(fù)雜，Jetty 8.x 比較簡單，代碼好讀一些～

QueuedThreadPool

QueuedThreadPool 類層次結(jié)構(gòu)

ThreadPool
    SizedThreadPool
        QueuedThreadPool

線程池核心問題：

線程管理：包括創(chuàng)建，銷毀等

任務(wù)分派：推或拉模型

線程管理

通過最小線程個數(shù)，最大線程個數(shù)，線程最大空閑時間來動態(tài)創(chuàng)建，銷毀線程

線程池初始化

// QueuedThreadPool.java
public class QueuedThreadPool {
    // 默認(rèn)最大線程數(shù)
    private int _maxThreads = 254;
    // 默認(rèn)最小線程數(shù)
    private int _minThreads = 8;
    
    ...

    @Override
    protected void doStart() throws Exception {
        ...
        int threads = _threadsStarted.get();
        // 啟動 _minThreads 個線程
        while (isRunning() && threads < _minThreads) {
            startThread(threads);
            threads = _threadStarted.get();
        }
    }
}

線程銷毀

當(dāng)線程空閑時間超過設(shè)定的閾值 _maxIdleTimeMs 而且線程池的線程個數(shù)高于 _minThreads 時，線程將從 Runnable run 方法中退出

# QueuedThreadPool.java
private Runnable _runnable = new Runnable() {
    public void run() {
        boolean shrink = false;
        ...
        try {
            Runnable job = _jobs.poll()
            while (isRunning()) {
                // job loop：從隊(duì)列中拉取 job 并執(zhí)行直到隊(duì)列為空
                while (job != null && isRunning()) {
                    runJob(job);
                    job = _jobs.poll();
                }

                // idle loop
                try {
                    _threadsIdle.incrementAndGet();
                    while (isRunning() && job == null) {
                        // 如果 _maxIdleTimeMs < 0 就進(jìn)入阻塞模式，直到成功拉取到 job
                        if (_maxIdleTimeMs <= 0) {
                            job = _jobs.take();
                        } else {
                            final int size = _threadsStarted.get();
                            // 如果線程個數(shù)少于 _minThreads 那么不需要回收線程
                            if (size > _minThreads) {
                                long last = _lastShrink.get();
                                long now = System.currentTimeMillis();
                                if (last == 0 || (now - last) > _maxIdleTimeMs) {
                                    shrink = _lastShrink.compareAndSet(last, now) &&
                                        _threadsStarted.compareAndSet(size, size - 1);
                                    if (shringk) {
                                        // 退出 while 循環(huán)，即終止線程
                                        return;
                                    }
                                }
                            }
                            // 帶 timeout 的 poll
                            job = idleJobPoll();
                        }
                    }
                }
            }
        } catch (...) {
        } finally {
        }
    }
}

任務(wù)分派

使用阻塞隊(duì)列，工作線程從阻塞隊(duì)列中主動拉取（poll）任務(wù)（job），QueuedThreadPool 默認(rèn)使用 ArrayBlockingQueue，如果需要可以通過構(gòu)造方法手動指定阻塞隊(duì)列的具體實(shí)現(xiàn)

public class QueuedThreadPool {
    public QueuedThreadPool(BlockingQueue jobQ) {
        this();
        _jobs = jobQ;
        _jobs.clear();
    }

    public boolean dispatch(Runnable job) {
        if (isRunning()) {
            final int jobQ = _jobs.size();
            final int idle = getIdleThreads();
            // 將 job 放入隊(duì)列中
            if (_jobs.offer(job)) {
                // 如果當(dāng)前沒有 idle 的線程，或者 隊(duì)列中堆積的 job 比 idle 線程多那么嘗試啟動新的線程
                if (idle == 0 || jobQ > idle) {
                    int threads = _threadsStarted.get();
                    if (threads < _maxThreads) {
                        startThread(threads);
                    }
                }
                return true;
            }
        }
        return false;
    }
}

啟動流程

Server 的啟動開始于 start 方法的調(diào)用，如上文所述，Server 類繼承自 AggregateServer，父類的 start 方法最終調(diào)用 Server 的 doStart 方法

Server start

@Override
protected void doStart() throws Exception {
    ...
    // 創(chuàng)建默認(rèn)的 ThreadPool
    if (_threadPool == null) {
        setThreadPool(new QueuedThreadPool())
    }
    try {
        super.doStart();
    } catch(Throwable e) {...}
    // 啟動 Connector，接收，處理請求
    if (_connectors != null && mex.size == 0) {
        for (int i = 0; i < _connectors.length; i++) {
            try {
                _connectors[i].start();
            } catch (Throwable e) {
                mex.add(e);
            }
        }
    }
    ...
}