前一段時間需要寫一個 HTTP Server，之前用 Akka 實現了一個，但是 Netty 更能支撐高併發，連 Spark 2.0 也把 Akka 換成了 Netty 進行網路傳輸，所以現在再擼一個 Netty 的 Server。

閒話少說，直接開搞，首先將 Netty 加入 sbt：

"io.netty" % "netty-all" % "4.1.25.Final"

我們需要定義兩個組 bossGroup 和 workerGroup，這個兩個組實際上是兩個執行緒組，負責的不同。bossGroup 負責獲取客戶端連線，接收到之後會將該連線轉發到 workerGroup 進行處理。

// if there are multiple `ServerBootstrap`, we should set more than one thread
val bossGroup = new NioEventLoopGroup(1)
val workerGroup = new NioEventLoopGroup()

然後通過一個啟動服務類 ServerBootstrap 進行初始化啟動：

val http = new ServerBootstrap()
  .group(bossGroup, workerGroup)
  .channel(classOf[NioServerSocketChannel])
  .childHandler(new
 
 HttpServerInitializer(writer))

其中 HttpServerInitializer 是我們自定義實現的一個服務初始化器，繼承了 ChannelInitializer 類，這個類通過 ChannelPipieline 設定初始化鏈，包括超時時間、編解碼和處理 HTTP 請求的設定。

class HttpServerInitializer(private val writer: EventWriter) extends ChannelInitializer[SocketChannel] {
  private val logger = Logger(this.getClass)

  override
 
 def initChannel(ch: SocketChannel) {
    logger.debug("Initialize http server channel.")
    ch.pipeline.addLast(
      new ReadTimeoutHandler(1),
      new HttpRequestDecoder(8192, 8192, 8192),
      new HttpResponseEncoder(),
      new HttpServerHandler(writer)
    )
  }
}

這裡的 HttpServerHandler 就是主要的 HTTP 請求處理類，類似於 Spring 中的攔截器，包含處理請求、轉發的相關功能：

class HttpServerHandler(private val writer: EventWriter)
  extends SimpleChannelInboundHandler[Object] {
  private val logger = Logger(this.getClass)

  private def forwardMessage(req: HttpRequest) {
    if (req.decoderResult.isSuccess) {
      val uri = req.uri
      logger.debug(s"Forward message, url=$uri")
      writer.putEvent(System.currentTimeMillis, uri)
    } else {
      logger.error(s"Request decode failure, result=${req.decoderResult}")
    }
  }

  override def channelReadComplete(ctx: ChannelHandlerContext) {
    ctx.flush
  }

  override def channelRead0(ctx: ChannelHandlerContext, msg: Object) {
    if (msg.isInstanceOf[HttpRequest]) {
      logger.debug("Flush and finish requests.")
      val response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NO_CONTENT, Unpooled.EMPTY_BUFFER)
      ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE)
      forwardMessage(msg.asInstanceOf[HttpRequest])
    }
  }

  override def exceptionCaught(ctx: ChannelHandlerContext, error: Throwable) {
    logger.error(s"Http handle error, error=$error")
    ctx.close
  }

}

注意這裡是通過 channelRead0 處理請求並給客戶端返回，我在這裡的處理是不論什麼內容進來，都給客戶端返回一個 NO_CONTENT，而實際的處理會進行轉發，這樣是為了防止阻塞，首先給客戶端返回結果，然後具體的一些耗時操作放在 forwardMessage 內。

上面的內容就是一些初始化啟動類，準備好這些以後我們需要把初始化伺服器繫結到服務埠，這個埠用來接收 HTTP 請求：

val f = http.bind(port).sync().channel()
logger.info(s"Start server, port=$port")
f.closeFuture().sync()

如果接收 HTTP 請求的時候出錯，我們還需要釋放資源：

bossGroup.shutdownGracefully()
workerGroup.shutdownGracefully()

以上完整實現了一個 HTTP Server，具體的過程是：

1. 定義兩個執行緒組： bossGroup 和 workerGroup
2. 準備一個繼承自 ChannelInitializer 服務處理初始化類，這個類設定超時編解碼以及具體處理 HTTP 請求的類
3. 定義一個服務啟動類 ServerBootstrap，設定執行緒組和 handler
4. 把服務啟動類繫結到某個埠號，HTTP Server 啟動

完整程式碼見 producer，我是通過 Netty 的 HTTP Server 接收 GET 請求，處理之後將訊息傳送到 AWS Kinesis（類似於 Kafka）中。