附錄C Jetty與JettyUtils

注：本文是為了配合《Spark核心設計的藝術架構設計與實作》一書的内容而編寫，目的是為了節省成本、友善讀者查閱。書中附錄C的内容都在本文呈現。

Jetty簡介

Jetty是一個開源的，以Java作為開發語言的servlet容器。它的API以一組JAR包的形式釋出。Jetty容器可以執行個體化成一個對象，因而迅速為一些獨立運作的Java應用提供網絡和web服務。要為Jetty建立servlet，就涉及ServletContextHandler的API使用。示例代碼如下：

class HelloServlet extends HttpServlet {  
  private static final long serialVersionUID = 1L;  
  private String msg = "Hello World!";  

  protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {  
    response.setContentType("text/html");  
    response.setStatus(HttpServletResponse.SC_OK);  
    response.getWriter().println("<h1>" + msg + "</h1>");  
    response.getWriter().println("session=" + request.getSession(true).getId());  
  }  
}
public static void main(String[] args) throws Exception {  
  Server server = new Server(8080);  
  ServletContextHandler context = new ServletContextHandler();  
  context.setContextPath("/");  
  server.setHandler(context);  
  // http://localhost:8080/hello  
  context.addServlet(new ServletHolder(new HelloServlet()), "/hello");
  server.start();  
  server.join();
}

如果想更深入了解Jetty，請通路官網

http://www.eclipse.org/jetty/

JettyUtils

JettyUtils是Spark對于Jetty相關API的又一層封裝，這裡對其中一些主要類型和方法進行介紹。

ServerInfo

功能描述：提供給Jetty伺服器添加或移除ContextHandler，以及停止Jetty伺服器的實作。

private[spark] case class ServerInfo(
    server: Server,
    boundPort: Int,
    securePort: Option[Int],
    private val rootHandler: ContextHandlerCollection) {

  def addHandler(handler: ContextHandler): Unit = {
    handler.setVirtualHosts(Array("@" + JettyUtils.SPARK_CONNECTOR_NAME))
    rootHandler.addHandler(handler)
    if (!handler.isStarted()) {
      handler.start()
    }
  }

  def removeHandler(handler: ContextHandler): Unit = {
    rootHandler.removeHandler(handler)
    if (handler.isStarted) {
      handler.stop()
    }
  }

  def stop(): Unit = {
    server.stop()
    // Stop the ThreadPool if it supports stop() method (through LifeCycle).
    // It is needed because stopping the Server won't stop the ThreadPool it uses.
    val threadPool = server.getThreadPool
    if (threadPool != null && threadPool.isInstanceOf[LifeCycle]) {
      threadPool.asInstanceOf[LifeCycle].stop
    }
  }
}

createServlet

功能描述：建立javax.servlet.http.HttpServlet的匿名内部類執行個體。此執行個體處理請求實際是使用servletParams的responder：Responder，此Responder類型發生隐式轉換，會轉換為使用者傳入的函數參數。

def createServlet[T <% AnyRef](
      servletParams: ServletParams[T],
      securityMgr: SecurityManager,
      conf: SparkConf): HttpServlet = {

    val allowFramingFrom = conf.getOption("spark.ui.allowFramingFrom")
    val xFrameOptionsValue =
      allowFramingFrom.map(uri => s"ALLOW-FROM $uri").getOrElse("SAMEORIGIN")

    new HttpServlet {
      override def doGet(request: HttpServletRequest, response: HttpServletResponse) {
        try {
          if (securityMgr.checkUIViewPermissions(request.getRemoteUser)) {
            response.setContentType("%s;charset=utf-8".format(servletParams.contentType))
            response.setStatus(HttpServletResponse.SC_OK)
            val result = servletParams.responder(request)
            response.setHeader("Cache-Control", "no-cache, no-store, must-revalidate")
            response.setHeader("X-Frame-Options", xFrameOptionsValue)
            response.getWriter.print(servletParams.extractFn(result))
          } else {
            response.setStatus(HttpServletResponse.SC_UNAUTHORIZED)
            response.setHeader("Cache-Control", "no-cache, no-store, must-revalidate")
            response.sendError(HttpServletResponse.SC_UNAUTHORIZED,
              "User is not authorized to access this page.")
          }
        } catch {
          case e: IllegalArgumentException =>
            response.sendError(HttpServletResponse.SC_BAD_REQUEST, e.getMessage)
          case e: Exception =>
            logWarning(s"GET ${request.getRequestURI} failed: $e", e)
            throw e
        }
      }
      // SPARK-5983 ensure TRACE is not supported
      protected override def doTrace(req: HttpServletRequest, res: HttpServletResponse): Unit = {
        res.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED)
      }
    }
  }

createServletHandler

功能描述：建立以給定路徑為字首的請求的ServletContextHandler。處理步驟如下：

1) 調用createServlet，生成javax.servlet.http.HttpServlet的匿名内部類執行個體。此執行個體處理請求實際是使用servletParams的responder：Responder，此Responder類型發生隐式轉換，會轉換為使用者傳入的函數參數。

2)調用重載的createServletHandler方法，生成org.eclipse.jetty.servlet.ServletHolder ，并最終生成ServletContextHandler。

createServletHandler的實作如下。

def createServletHandler[T <% AnyRef](
      path: String,
      servletParams: ServletParams[T],
      securityMgr: SecurityManager,
      conf: SparkConf,
      basePath: String = ""): ServletContextHandler = {
    createServletHandler(path, createServlet(servletParams, securityMgr, conf), basePath)
  }

  /** Create a context handler that responds to a request with the given path prefix */
  def createServletHandler(
      path: String,
      servlet: HttpServlet,
      basePath: String): ServletContextHandler = {
    val prefixedPath = if (basePath == "" && path == "/") {
      path
    } else {
      (basePath + path).stripSuffix("/")
    }
    val contextHandler = new ServletContextHandler
    val holder = new ServletHolder(servlet)
    contextHandler.setContextPath(prefixedPath)
    contextHandler.addServlet(holder, "/")
    contextHandler
  }

createStaticHandler

功能描述：建立對靜态目錄提供檔案服務的ServletContextHandler。

def createStaticHandler(resourceBase: String, path: String): ServletContextHandler = {
    val contextHandler = new ServletContextHandler
    contextHandler.setInitParameter("org.eclipse.jetty.servlet.Default.gzip", "false")
    val staticHandler = new DefaultServlet
    val holder = new ServletHolder(staticHandler)
    Option(Utils.getSparkClassLoader.getResource(resourceBase)) match {
      case Some(res) =>
        holder.setInitParameter("resourceBase", res.toString)
      case None =>
        throw new Exception("Could not find resource path for Web UI: " + resourceBase)
    }
    contextHandler.setContextPath(path)
    contextHandler.addServlet(holder, "/")
    contextHandler
  }

createRedirectHandler

功能描述：建立将使用者對源路徑的請求總是重定向到目标路徑的ServletContextHandler。

def createRedirectHandler(
      srcPath: String,
      destPath: String,
      beforeRedirect: HttpServletRequest => Unit = x => (),
      basePath: String = "",
      httpMethods: Set[String] = Set("GET")): ServletContextHandler = {
    val prefixedDestPath = basePath + destPath
    val servlet = new HttpServlet {
      override def doGet(request: HttpServletRequest, response: HttpServletResponse): Unit = {
        if (httpMethods.contains("GET")) {
          doRequest(request, response)
        } else {
          response.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED)
        }
      }
      override def doPost(request: HttpServletRequest, response: HttpServletResponse): Unit = {
        if (httpMethods.contains("POST")) {
          doRequest(request, response)
        } else {
          response.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED)
        }
      }
      private def doRequest(request: HttpServletRequest, response: HttpServletResponse): Unit = {
        beforeRedirect(request)
        // Make sure we don't end up with "//" in the middle
        val newUrl = new URL(new URL(request.getRequestURL.toString), prefixedDestPath).toString
        response.sendRedirect(newUrl)
      }
      // SPARK-5983 ensure TRACE is not supported
      protected override def doTrace(req: HttpServletRequest, res: HttpServletResponse): Unit = {
        res.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED)
      }
    }
    createServletHandler(srcPath, servlet, basePath)
  }

startJettyServer

功能描述：建立以給定路徑為字首的請求的響應處理。處理步驟如下：

1) 将SparkUI中的全部handler加入ContextHandlerCollection。

2) 如果使用配置spark.ui.filters指定了filter，則給所有handler增加filter。

3) 調用Utils的方法startServiceOnPort，最終回調函數connect。

startJettyServer的實作如下。

def startJettyServer(
      hostName: String,
      port: Int,
      sslOptions: SSLOptions,
      handlers: Seq[ServletContextHandler],
      conf: SparkConf,
      serverName: String = ""): ServerInfo = {

    addFilters(handlers, conf)

    val gzipHandlers = handlers.map { h =>
      h.setVirtualHosts(Array("@" + SPARK_CONNECTOR_NAME))

      val gzipHandler = new GzipHandler
      gzipHandler.setHandler(h)
      gzipHandler
    }

    // Bind to the given port, or throw a java.net.BindException if the port is occupied
    def connect(currentPort: Int): ((Server, Option[Int]), Int) = {
      val pool = new QueuedThreadPool
      if (serverName.nonEmpty) {
        pool.setName(serverName)
      }
      pool.setDaemon(true)

      val server = new Server(pool)
      val connectors = new ArrayBuffer[ServerConnector]()
      val collection = new ContextHandlerCollection

      // Create a connector on port currentPort to listen for HTTP requests
      val httpConnector = new ServerConnector(
        server,
        null,
        // Call this full constructor to set this, which forces daemon threads:
        new ScheduledExecutorScheduler(s"$serverName-JettyScheduler", true),
        null,
        -1,
        -1,
        new HttpConnectionFactory())
      httpConnector.setPort(currentPort)
      connectors += httpConnector

      val httpsConnector = sslOptions.createJettySslContextFactory() match {
        case Some(factory) =>
          // If the new port wraps around, do not try a privileged port.
          val securePort =
            if (currentPort != 0) {
              (currentPort + 400 - 1024) % (65536 - 1024) + 1024
            } else {
              0
            }
          val scheme = "https"
          // Create a connector on port securePort to listen for HTTPS requests
          val connector = new ServerConnector(server, factory)
          connector.setPort(securePort)
          connector.setName(SPARK_CONNECTOR_NAME)
          connectors += connector

          // redirect the HTTP requests to HTTPS port
          httpConnector.setName(REDIRECT_CONNECTOR_NAME)
          collection.addHandler(createRedirectHttpsHandler(securePort, scheme))
          Some(connector)

        case None =>
          // No SSL, so the HTTP connector becomes the official one where all contexts bind.
          httpConnector.setName(SPARK_CONNECTOR_NAME)
          None
      }

      // As each acceptor and each selector will use one thread, the number of threads should at
      // least be the number of acceptors and selectors plus 1. (See SPARK-13776)
      var minThreads = 1
      connectors.foreach { connector =>
        // Currently we only use "SelectChannelConnector"
        // Limit the max acceptor number to 8 so that we don't waste a lot of threads
        connector.setAcceptQueueSize(math.min(connector.getAcceptors, 8))
        connector.setHost(hostName)
        // The number of selectors always equals to the number of acceptors
        minThreads += connector.getAcceptors * 2
      }
      pool.setMaxThreads(math.max(pool.getMaxThreads, minThreads))

      val errorHandler = new ErrorHandler()
      errorHandler.setShowStacks(true)
      errorHandler.setServer(server)
      server.addBean(errorHandler)

      gzipHandlers.foreach(collection.addHandler)
      server.setHandler(collection)

      server.setConnectors(connectors.toArray)
      try {
        server.start()
        ((server, httpsConnector.map(_.getLocalPort())), httpConnector.getLocalPort)
      } catch {
        case e: Exception =>
          server.stop()
          pool.stop()
          throw e
      }
    }

    val ((server, securePort), boundPort) = Utils.startServiceOnPort(port, connect, conf,
      serverName)
    ServerInfo(server, boundPort, securePort,
      server.getHandler().asInstanceOf[ContextHandlerCollection])
  }

關于《Spark核心設計的藝術架構設計與實作》

經過近一年的準備，《Spark核心設計的藝術架構設計與實作》一書現已出版發行，圖書如圖：

紙質版售賣連結如下：

京東：

https://item.jd.com/12302500.html

電子版售賣連結如下：

https://e.jd.com/30389208.html

附錄C Jetty與JettyUtils

Jetty簡介

JettyUtils

ServerInfo

createServlet

createServletHandler

createStaticHandler

createRedirectHandler

startJettyServer

關于《Spark核心設計的藝術架構設計與實作》

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method

附錄C Jetty與JettyUtils

Jetty簡介

JettyUtils

ServerInfo

createServlet

createServletHandler

createStaticHandler

createRedirectHandler

startJettyServer

關于《Spark核心設計的藝術 架構設計與實作》

繼續閱讀

關于《Spark核心設計的藝術架構設計與實作》