Akka實戰：構建REST風格的微服務

使用Akka-Http構建REST風格的微服務，服務API應儘可能遵循REST語義，數據使用JSON格式交互。在有錯誤發生時應返回：{"errcode":409,"errmsg":"aa is invalid，the ID is expected to be bb"}相似的JSON錯誤消息。

代碼：

• https://github.com/yangbajing/akka-action
• http://git.oschina.net/yangbajing/akka-action

代碼

首先來看看代碼文件結構：

├── ApiRoute.scala
├── App.scala
├── ContextProps.scala
├── book
│   ├── Book.scala
│   ├── BookContextProps.scala
│   ├── BookRoute.scala
│   └── BookService.scala
└── news
    ├── News.scala
    ├── NewsContextProps.scala
    ├── NewsRoute.scala
    └── NewsService.scala

經過名字能夠看出，App.scala是啓動程序，以Route結尾的是API路由定義文件，Service結尾的就是服務實現代碼了。ContextProps結尾的是服務與路由交互的上下文屬性部分，Service的將會在ContextProps中實例化並傳給各個Route。

從目錄結構上看，程序是按功能模塊進行劃分的。book相關的路由、服務、實體都定義在book包下。相應的，與news相關的代碼則寫於news包。

首先來看看程序的啓動文件，App.scala

def main(args: Array[String]): Unit = {
    Files.write(Paths.get("app.pid"), Utils.getPid.getBytes(Utils.CHARSET))

    val contextProps = new ContextProps

    val bindingFuture = Http().bindAndHandle(ApiRoute(contextProps), "0.0.0.0", 3333)

    bindingFuture.onComplete {
      case Success(binding) =>
        logger.info(binding.toString)
      case Failure(e) =>
        logger.error(e.getLocalizedMessage, e)
    }
  }

定義akka-http綁定的host和port，設置ContextProps，並把它傳給ApiRoute。App.scala的代碼仍是很簡單的，接下來看看ApiRoute的實現。

// 定義一個Health Check API，用戶第3方工具（如：Nginx/Tengine）驗證服務是否正常運行
  val healthCheck =
    path("health_check") {
      get { ctx =>
        logger.debug(ctx.request.toString)
        ctx.complete(HttpEntity.Empty)
      }
    }

  import me.yangbajing.akkaaction.util.JsonSupport._

  val customExceptionHandler = ExceptionHandler {
    case e: MessageException =>
      extractRequest { req =>
        val msg =
          s"""\nmethod: ${req.method}
             |uri: ${req.uri}
             |headers:
             |\t${req.headers.mkString("\n\t")}
             |$e""".stripMargin
        if (e.errcode > 500) logger.error(msg, e)
        else logger.warn(msg)

        complete(
          StatusCodes.getForKey(e.errcode) getOrElse StatusCodes.InternalServerError,
          JObject("errcode" -> JInt(e.errcode), "errmsg" -> JString(e.errmsg)))
      }
    case e: Exception =>
      extractRequest { req =>
        logger.error(req.toString, e)
        complete(
          StatusCodes.InternalServerError,
          JObject("errcode" -> JInt(500), "errmsg" -> JString(e.getLocalizedMessage)))
      }
  }

  def apply(props: ContextProps)(implicit ec: ExecutionContextExecutor, mat: Materializer) =
    handleExceptions(customExceptionHandler) {
      pathPrefix("api") {
        healthCheck ~
          NewsRoute(props) ~
          BookRoute(props)
      }
    }

代碼有一點長，如今分別解說。

customExceptionHandler

自定義的異常處理器，主要用於把自定義異常和系統異常轉換成JSON消息輸出，並設置相對應的HTTP狀態碼。

apply

apply方法定義了實現的API路由，由代碼能夠看到news、book兩個模塊的路由分別由NewsRoute和BookRoute兩個文件定義。把相同功能的路由、服務、實體定義在同一個邏輯上下文（包）中，我的認爲是一種更好的微服務實踐。

book模塊詳解

book
├── Book.scala
├── BookContextProps.scala
├── BookRoute.scala
└── BookService.scala

• Book：實體
• BookContextProps：上下文屬性，服務將在此實例化。並把接口混入ContextProps中。
• BookRoute：API路由定義
• BookService：服務功能實現

BookRotue定義

pathPrefix("book") {
      pathEnd {
        post {
          entity(as[Book]) { book =>
            onSuccess(props.bookService.persist(book)) { result =>
              complete(StatusCodes.Created, result)
            }
          }
        }
      } ~
        path(Segment) { bookId =>
          get {
            complete(props.bookService.findOneById(bookId))
          } ~
            put {
              entity(as[Book]) { book =>
                complete(props.bookService.updateById(bookId, book))
              }
            } ~
            delete {
              complete(props.bookService.deleteById(bookId).map(id => Map("id" -> id)))
            }
        }

Akka-Http提供了高級routing DSL，能夠很天然的定義出樹型結構的RESTful風格的API。由代碼可見，定義了4個API。分別對應insert、select、update、delete操做，由post、get、put和delete4個指令實現對應操做的HTTP方法。

pathPrefix、pathEnd和path3個路徑定義指令的區別在於pathPrefix表明由它定義的路徑還並未終結，在它下面還有子路徑。而path則表明它已是最終的路徑了，pathEnd是用於在使用了pathPrefix的狀況下也能夠直接訪問由pathPrefix指定的路徑。

Segment用於把由path定義的路徑抽取成一個參數（bookId）。除了Segment用於抽取一個字符串類型，還有IntNumber和LongNumber用於抽取路徑爲Int或Long類型。

entity指令用於抽取HTTP請求的body部分，並經過as[T]方法將其自動解析爲指定類型。這裏使用到了akka提供的Unmarshaller特性。這裏經過JsonSupport裏定義的json4sUnmarshaller將用戶請求提交的JSON字符串映射到Book類型。

implicit def json4sUnmarshaller[A: Manifest](implicit mat: Materializer): FromEntityUnmarshaller[A] =
    Unmarshaller.byteStringUnmarshaller
      .forContentTypes(MediaTypes.`application/json`)
      .mapWithCharset { (data, charset) =>
        val input = if (charset == HttpCharsets.`UTF-8`) data.utf8String else data.decodeString(charset.nioCharset().name)
        jsonSerialization.read(input)
      }

  implicit def json4sMarshaller[A <: AnyRef]: ToEntityMarshaller[A] =
    Marshaller.StringMarshaller.wrap(ContentType(MediaTypes.`application/json`, HttpCharsets.`UTF-8`))(v =>
      jsonSerialization.write[A](v))

天然json4sMarshaller則是把T類型的對象映射爲JSON字符串響應給請求方。

BookService

再來看看BookService服務的實現。

def updateById(bookId: String, book: Book)(implicit ec: ExecutionContext): Future[Book] = Future {
    if (bookId != book.id)
      throw MeConflictMessage(s"${book.id} is invalid，the ID is expected to be $bookId")

    val newBooks = BookService.books.filterNot(_.id == bookId)
    if (newBooks.size == BookService.books.size)
      throw MeNotFoundMessage(s"$bookId not found")

    BookService.books ::= book
    book
  }

  def persist(book: Book)(implicit ec: ExecutionContext): Future[Book] = Future {
    if (BookService.books.exists(_.id == book.id))
      throw MeConflictMessage(s"${book.id} exsits")

    BookService.books ::= book
    book
  }

  def deleteById(bookId: String)(implicit ec: ExecutionContext): Future[String] = Future {
    val newBooks = BookService.books.filterNot(_.id == bookId)
    if (newBooks.size == BookService.books.size)
      throw MeNotFoundMessage(s"$bookId not found")

    BookService.books = newBooks
    bookId
  }

  def findOneById(bookId: String)(implicit ec: ExecutionContext): Future[Book] = Future {
    BookService.books.find(_.id == bookId).getOrElse(throw MeNotFoundMessage(s"$bookId not found"))
  }

看到每一個方法的返回值都被定義成了Future[T]，akka-http是一個基於akka-actor和akka-stream的異步HTTP工具集，使用Future能夠提供整個系統的響應。咱們這裏直接使用Future來模擬異步訪問（數據庫操做）。

在每一個方法中，咱們校驗參數是否有效。若校驗失敗則直接拋出自定義異常。Future函數將捕獲異常，由以前定義的customExceptionHandler自定義異常處理器來將自定義異常轉換成JSON消息發送給調用方，並設置匹配的HTTP狀態碼。

測試

百聞不如一試，下載代碼實際操做下（下載地址在文章開頭）。

運行程序：

./sbt
akka-action > runMain me.yangbajing.akkaaction.restapi.App

依次執行docs/scripts/restapi目錄下的測試腳本，查看各請求下REST API的返回值（須要系統安裝curl）。

• ./get-book-aa.sh：正常返回ID爲aa的書
• ./get-book-bb.sh：查找ID爲bb的書返回404
• ./post-book.sh：建立一本ID爲bb的書，返回201
• ./get-book-bb.sh：正確返回ID爲bb的書
• ./put-book.hs：正確更新ID爲bb的書
• ./put-book-invalid.sh：無效的更新ID爲aa的書，返回409
• ./delete-book-aa.sh：成功的刪除ID爲aa的書
• ./get-book-aa.sh：再次查找ID爲aa的書返回404
• ./delete-book-aa.sh：再次刪除ID爲aa的書時返回404

總結

akka-http是一個頗有意思的HTTP工具庫，它完整的實現了客戶端和服務端編程工具，還支持WebScoket。基於akka-actor和akka-stream，提供了高併發的異步編程模型。咱們能夠很快捷的實現出一個響應式（Reactive）Web Service。其提供的routing DSL可方便的定義出一套樹型結構的API，很天然的匹配到RESTful風格的API設計。