使用Akka HTTP构建微服务：CDC方法

欢迎大家前往腾讯云+社区，获取更多腾讯海量技术实践干货哦~

本文来自云+社区翻译社，作者阿小庆

构建微服务并不容易，特别是当微服务变得越来越多时，而且好多微服务可能由不同的团队提供和维护，这些微服务彼此交互并且变化很快。

文档、团队交互和测试是获得成功的三大法宝，但是如果用错误的方式进行，它们会产生更多的复杂性，而不是一种优势。

我们可以使用像Swagger（用于文档），Docker（用于测试环境），Selenium（用于端到端测试）等工具，但是我们最终还是会因为更改API而浪费大量时间，因为他们不是说谁适合来使用它们，或者设置合适的环境来执行集成测试，而是需要生产数据（希望是匿名的），但生产数据可能需要很长时间才能完成。

对所有这些问题都没有正确的答案，但我认为有一件事可以帮助很多人：首先从用户角度出发！

这是什么意思？一般情况下，在开发Web应用程序的时候，从模型和流程定义开始，深入到软件开发中，都是使用TDD（测试驱动开发）方法：先写测试，考虑我们真正想要的，以及我们如何使用它; 但微服务（microservices）呢？在这种情况下，它从消费者开始！消费者希望从其他服务中获得什么以及它希望如何互动？

这就是我说的 消费者驱动的契约（CDC） 测试。采用这种方法，消费者自己会定义需要的数据格式以及交互细节，并驱动生成一份契约文件。然后生产者根据契约文件来实现自己的逻辑，并在持续集成环境中持续验证。

商业案例

比如，我们希望在“我的图书馆”实现一项新功能，所以我们需要介绍类别（Categories），并且我们想知道其中有多少类别。这个想法是将逻辑分成两个服务，一个生产者（Producer）提供所有类别的列表，另一个消费者（Consumer）对其进行计数。

非常容易，但足以创建一个良好的基础结构和对CDC的理解。

技术栈

这篇文章，我选择了Scala作为语言，Akka HTTP作为框架。我认为这是一项非常好的技术，它可以满足构建微服务所需的所有基本要求：

• 易于实现
• 快速
• 健壮性
• 很好的支持和文档记录

在数据方面，我选择了Slick作为库，将数据库交互和FlyWay抽象为数据库迁移框架。它们既健壮又稳定，多次使用也没有问题。

最后，也是很重要的一点，测试支持！我喜欢Scala Test，因为它始终是我在Scala的项目的一部分，但我们的CDC呢？

对于CDC，有一个非常好的框架，可用于多平台：Pact。

通过Pact，我们可以定义我们的消费者契约文件，并根据微服务接口的提供者和消费者进行验证。我建议花几分钟阅读官方Pact网站的主页，这很好地诠释了它背后的道理。

正如我所说的，Pact适用于很多平台，在我们的例子中，用Scala编写Consumer和Producer，我们只能使用一个实现： Scala-Pact 。

操作

为了简单起见，我已经创建了一个包含消费者和生产者的SBT项目，但它们可以很容易被分割并用作模板。你可以在 github.com/mariniss/my… 找到源代码。

让我们以CDC风格开始我们的微服务实现！首先，我们必须定义我们的项目。我们可以轻松地使用SBT创建一个新的Scala项目并定义build.sbt，如下所示：

build.sbt

name := "myLibrary-contracts"
version := "0.1"
scalaVersion := "2.12.4"
enablePlugins(ScalaPactPlugin)
libraryDependencies ++= Seq(
 //Common dependencies
 "com.typesafe.akka"  %% "akka-stream"             % "2.4.20",
 "com.typesafe.akka"  %% "akka-http"               % "10.0.11", // Akka HTTP项目的标准依赖关系
 "com.typesafe.akka"  %% "akka-http-spray-json"    % "10.0.11", // 用于JSON序列化和反序列化
 "org.slf4j"           % "slf4j-simple"            % "1.7.25",  // 用于日志记录
 "org.scalatest"      %% "scalatest"               % "3.0.1"   % "test",  // 测试框架
 "org.scalamock"      %% "scalamock"               % "4.0.0"   % "test",  // 模拟框架
 "com.typesafe.akka"  %% "akka-stream-testkit"     % "2.4.20"  % "test",
 "com.typesafe.akka"  %% "akka-testkit"            % "2.4.20"  % "test",
 "com.typesafe.akka"  %% "akka-http-testkit"       % "10.0.11" % "test",
 "com.itv"            %% "scalapact-argonaut-6-2"  % "2.2.0"   % "test",
 "com.itv"            %% "scalapact-scalatest"     % "2.2.0"   % "test",
 "com.itv"            %% "scalapact-http4s-0-16-2" % "2.2.0"   % "test",
 //Producer dependencies
 "com.typesafe.slick" %% "slick"                   % "3.2.1",
 "com.typesafe.slick" %% "slick-hikaricp"          % "3.2.1",
 "com.h2database"      % "h2"                      % "1.4.196",
 "org.flywaydb"        % "flyway-core"             % "5.0.7"
)
testOptions in Test += Tests.Argument(TestFrameworks.ScalaTest,
 "-y", "org.scalatest.WordSpec",
 "-y", "org.scalatest.FunSpec")
parallelExecution in Test := false

正如你所看到的，Akka HTTP项目的标准依赖关系（通用于提供者和消费者）， spry-json 用于JSON序列化和反序列化，SL4J用于日志记录，scalatest和scalamock作为测试和模拟框架，以及Scala协议为CDC测试。

生产者特定的依赖关系仅用于数据库支持，如您所见，我使用H2（在内存数据库中），但您可以轻松地将其替换为其他数据库支持。

测试环境也有特定的配置; 只是因为我们在同一个项目中同时拥有生产者和客户端，所以并行执行被禁用，所以如果并行执行（我们稍后会看到它），我们可能会在Pact文件生成和使用过程中遇到问题。另外，我已经用两种不同的格式实现了测试，WordSpec和FunSpec，第一次用于所有的单元测试，第二次用于Pact测试，你可以按你的想法随意使用。

消费者（Consumer）操作

现在我们有了基本的项目结构，我们可以开始在消费者方面创建Pact测试，所以我们可以定义我们在给定特定场景/状态时对提供者（Provider）的期望。

*MyLibraryClientPactSpec.scala*

package com.fm.mylibrary.consumer.pact
import com.fm.mylibrary.consumer.MyLibraryClient
import com.fm.mylibrary.model.Category
import com.fm.mylibrary.model.JsonProtocol._
import com.itv.scalapact.ScalaPactForger._
import org.scalatest.{FunSpec, Matchers}
import spray.json._
class MyLibraryClientPactSpec extends FunSpec with Matchers {
 describe("Connecting to the MyLibrary server") {
   it("should be able to fetch the categories"){
     val categories = List(Category("Java"), Category("DevOps"))
     forgePact
       .between("ScalaConsumer")
       .and("myLibraryServer")
       .addInteraction(
         interaction
           .description("Fetching categories")
           .given("Categories: [Java, DevOps]")
           .uponReceiving(
             method = GET,
             path = "/search/category",
             query = None)
           .willRespondWith(
             status = 200,
             headers = Map("Content-Type" -> "application/json"),
             body = categories.toJson.toString())
       )
       .runConsumerTest { mockConfig =>
         val results = new MyLibraryClient().fetchCategories()
         results.isDefined shouldEqual true
         results.get.size shouldEqual 2
         results.get.forall(c => categories.contains(c)) shouldEqual true
       }
   }
 }
}

Scala-pact非常易于使用，这要归功于ScalaPactForger对象，可以通过几行代码构建契约定义和期望效果，更详细地说：

• 契约参与者的定义： .between("ScalaConsumer") .and("myLibraryServer")
• 参与者之间的相互作用的定义：

.addInteraction(interaction 
.description("Fetching categories") 
.given("Categories: [Java, DevOps]")
.uponReceiving(method = GET,path = "/search/category",query = None)
.willRespondWith( status = 200, headers = Map("Content-Type" -> "application/json"),body = categories.toJson.toString()))givenuponReceivingwillRespondWith

真正重要的是描述系统状态，其中交互必须如所描述的那样工作，由消费者uponReceiving执行的请求和预期的响应。同时考虑到所有HTTP元素必须匹配（方法，url，标题，正文和查询）

• 用于验证消费者契约的实际测试的定义： 此代码将针对以前的方案运行，虚拟服务器将响应 交互部分中定义的唯一HTTP请求（如果响应为deined），它将验证消费者（Consumer）是否将按照协议中的规定进行要求。也可以在消费者（Consumer）处理的结果值上添加更多的检查（声明）。

.runConsumerTest { mockConfig =>

val results = new MyLibraryClient().fetchCategories()

results.isDefined shouldEqual true

results.get.size shouldEqual 2

results.get.forall(c => categories.contains(c)) shouldEqual true}

当然，我们可以添加更多场景和交互。我们也可以为许多生产者定义更多的契约。我建议通过“基本路径”和标准错误情景来确定描述正常使用情况下所需的基本情景和交互情况，但是留给单元测试所有详细的测试，以及与它们的实现相关的各种情况。

现在，您可以尝试编译并执行测试，但由于我们没有客户端和模型，所以我们需要添加基本逻辑来让测试通过。

我认为我们可以通过两种方式进行，直接构建客户端（因为我们已经进行了测试），或者改进我们客户端的定义，创建单元测试并以纯TDD方式对其进行处理。我们来看第二个选项：

MyLibraryClientSpec.scala

package com.fm.mylibrary.consumer
import akka.http.scaladsl.model._
import com.fm.mylibrary.model.Category
import scala.concurrent.Future
class MyLibraryClientSpec extends BaseTestAppClient {
 implicit val myLibraryServerUrl:String = "//test"
 "Fetch categories" must {
   "execute the HTTP request to get all categories and returns them" in {
     val request = HttpRequest(HttpMethods.GET, "//test/search/category")
     val responseEntity = HttpEntity(bytes = """[{"name": "Java"}, {"name": "DevOps"}]""".getBytes,
                                     contentType = ContentTypes.`application/json`)
     val response = HttpResponse(status = StatusCodes.OK, entity = responseEntity)
     requestExecutor.expects(request).returning(Future.successful(response))
     val results = new MyLibraryClient().fetchCategories()
     results.isDefined shouldEqual true
     results.get.size shouldEqual 2
     results.get.contains(Category("Java")) shouldEqual true
     results.get.contains(Category("DevOps")) shouldEqual true
   }
 }
}

非常标准的测试; 我们希望抛出一个MyLibraryClient函数，该函数使用一个外部函数返回一个“Category”对象列表，该函数接受一个HttpRequest并返回一个HttpResponse。

正如你所看到的，没有明确提供这种外部依赖; 那是因为我想把它作为一个“隐含”价值。这是一种帮助创建可测试代码的方法，但 我强烈建议不要使用它 ，因为它会使代码难以阅读，特别是对于那些新的Scala。

我也喜欢定义一个具有所有必要依赖项的特征来轻松构建测试用例：

BaseTestAppClient.scala

package com.fm.mylibrary.consumer
import akka.actor.ActorSystem
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.ActorMaterializer
import akka.testkit.{ImplicitSender, TestKit}
import org.scalamock.scalatest.MockFactory
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import scala.concurrent.{ExecutionContextExecutor, Future}
class BaseTestAppClient extends TestKit(ActorSystem("BaseTestAppClient"))
           with WordSpecLike
           with ImplicitSender
           with Matchers
           with BeforeAndAfterAll
           with MockFactory {
 implicit val actorSystem: ActorSystem = system
 implicit val materializer: ActorMaterializer = ActorMaterializer()(system)
 implicit val executionContext: ExecutionContextExecutor = system.dispatcher
 implicit val requestExecutor = mockFunction[HttpRequest, Future[HttpResponse]]
 override def afterAll {
   TestKit.shutdownActorSystem(system)
 }
}

它定义了在我们的测试中使用的actor系统和执行HTTP请求的函数。

现在我们有了测试，让我们来实现一些逻辑：

MyClientLibrary.scala

package com.fm.mylibrary.consumer
import akka.actor.ActorSystem
import akka.http.scaladsl.client.RequestBuilding
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCodes}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import com.fm.mylibrary.model.Category
import com.fm.mylibrary.model.JsonProtocol._
import scala.concurrent.{ExecutionContextExecutor, Future}
class MyLibraryClient(implicit val myLibraryServerUrl: String,
                      implicit val actorSystem: ActorSystem,
                      implicit val materializer: ActorMaterializer,
                      implicit val executionContext: ExecutionContextExecutor,
                      implicit val requestExecutor: HttpRequest => Future[HttpResponse]) extends BaseHttpClient {
 def fetchCategories(): Option[List[Category]] = executeSyncRequest(
   RequestBuilding.Get(s"$myLibraryServerUrl/search/category"),
   response =>
     if(response.status == StatusCodes.OK)
       Unmarshal(response.entity).to[Option[List[Category]]]
     else
       Future.successful(None)
 )
}

Category.scala

package com.fm.mylibrary.model
case class Category (name: String)

这个相对容易实现。并且我使用了隐式声明依赖关系，但可以显性地提高代码的可读性。

接下来我创建了一个特征，它为每个HTTP客户端（现在只有一个）定义了基本组件，并具有一个以同步方式执行HTTP请求的功能：

BaseHttpClient.scala

package com.fm.mylibrary.consumer
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.ActorMaterializer
import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContextExecutor, Future}
import scala.language.postfixOps
trait BaseHttpClient {
 implicit def actorSystem: ActorSystem
 implicit def materializer: ActorMaterializer
 implicit def executionContext: ExecutionContextExecutor
 implicit def requestExecutor: HttpRequest => Future[HttpResponse]
 val awaitTime: FiniteDuration = 5000 millis
 def executeSyncRequest[T](request: HttpRequest, responseHandler: HttpResponse => Future[T]): T = {
   val response: Future[T] = requestExecutor(request).flatMap({ response =>
     responseHandler(response)
   })
   Await.result(response, awaitTime)
 }
}

现在我们很好地执行单元测试，如果我们没有犯错误，我们应该得到一个成功的执行。随意添加更多测试并重构客户端以便根据您的喜好调整结构（您可以 在此处 找到更多测试）。

我们也可以尝试执行Pact test（MyLibraryClientPactSpec），但它会失败，因为它应该执行一个真正的HTTP调用，scala-pact框架将启动一个真实的HTTP服务器，接受和响应协议中描述的请求。

我们差不多完成了我们想要的实现，它基本上是定义了actor系统和执行HTTP调用的函数的元素：

MyLibraryAppClient.scala

package com.fm.mylibrary.consumer.app
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.ActorMaterializer
import scala.concurrent.{ExecutionContextExecutor, Future}
object MyLibraryAppClient {
 implicit val actorSystem: ActorSystem = ActorSystem()
 implicit val materializer: ActorMaterializer = ActorMaterializer()
 implicit val executionContext: ExecutionContextExecutor = actorSystem.dispatcher
 implicit val requestExecutor: HttpRequest => Future[HttpResponse] = Http().singleRequest(_)
}

它是一个对象，所以我们可以将它导入到任何我们必须使用我们的客户端的地方，正如您在Pact测试中看到的那样： import com.fm.mylibrary.consumer.app.MyLibraryAppClient._

当然，您可以使用其他方法，但请**在选择时保持一致，**并避免在相同或类似项目中使用不同的方法/结构。

我们终于可以执行协议测试了！如果你很幸运，你应该得到这样的输出：

> Adding interactions:
> - Interaction(None,Some(Categories: [Java, DevOps]),Fetching categories,InteractionRequest(Some(GET),Some(/search/category),None,None,None,None),InteractionResponse(Some(200),Some(Map(Content-Type -> application/json)),Some([{"name":"Java"},{"name":"DevOps"}]),None))
[ScalaTest-run-running-MyLibraryClientPactSpec] INFO org.http4s.blaze.channel.nio1.NIO1SocketServerGroup - Service bound to address /127.0.0.1:55653
> ScalaPact stub running at: http://localhost:55653
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.ServerChannelGroup - Connection to /127.0.0.1:55666 accepted at Tue Feb 13 11:43:08 GMT 2018.
[http4s-blaze-client-1] INFO org.http4s.client.PoolManager - Shutting down connection pool: allocated=1 idleQueue.size=1 waitQueue.size=0
[DEBUG] [02/13/2018 11:43:09.376] [ScalaTest-run-running-MyLibraryClientPactSpec] [EventStream(akka://default)] logger log1-Logging$DefaultLogger started
[DEBUG] [02/13/2018 11:43:09.377] [ScalaTest-run-running-MyLibraryClientPactSpec] [EventStream(akka://default)] Default Loggers started
[DEBUG] [02/13/2018 11:43:09.595] [ScalaTest-run-running-MyLibraryClientPactSpec] [AkkaSSLConfig(akka://default)] Initializing AkkaSSLConfig extension...
[DEBUG] [02/13/2018 11:43:09.598] [ScalaTest-run-running-MyLibraryClientPactSpec] [AkkaSSLConfig(akka://default)] buildHostnameVerifier: created hostname verifier: com.typesafe.sslconfig.ssl.DefaultHostnameVerifier@db2cd5
[DEBUG] [02/13/2018 11:43:09.834] [default-akka.actor.default-dispatcher-5] [default/Pool(shared->http://localhost:55653)] (Re-)starting host connection pool to localhost:55653
[DEBUG] [02/13/2018 11:43:10.123] [default-akka.actor.default-dispatcher-5] [default/Pool(shared->http://localhost:55653)] InputBuffer (max-open-requests = 32) now filled with 1 request after enqueuing GET /search/category Empty
[DEBUG] [02/13/2018 11:43:10.127] [default-akka.actor.default-dispatcher-2] [default/Pool(shared->http://localhost:55653)] [0] Unconnected -> Loaded(1)
[DEBUG] [02/13/2018 11:43:10.137] [default-akka.actor.default-dispatcher-2] [default/Pool(shared->http://localhost:55653)] [0] <unconnected> Establishing connection...
[DEBUG] [02/13/2018 11:43:10.167] [default-akka.actor.default-dispatcher-2] [default/Pool(shared->http://localhost:55653)] [0] <unconnected> pushing request to connection: GET /search/category Empty
[DEBUG] [02/13/2018 11:43:10.179] [default-akka.actor.default-dispatcher-2] [akka://default/system/IO-TCP/selectors/$a/0] Resolving localhost before connecting
[DEBUG] [02/13/2018 11:43:10.200] [default-akka.actor.default-dispatcher-5] [akka://default/system/IO-DNS] Resolution request for localhost from Actor[akka://default/system/IO-TCP/selectors/$a/0#871918912]
[DEBUG] [02/13/2018 11:43:10.209] [default-akka.actor.default-dispatcher-5] [akka://default/system/IO-TCP/selectors/$a/0] Attempting connection to [localhost/127.0.0.1:55653]
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.ServerChannelGroup - Connection to /127.0.0.1:55669 accepted at Tue Feb 13 11:43:10 GMT 2018.
[DEBUG] [02/13/2018 11:43:10.212] [default-akka.actor.default-dispatcher-5] [akka://default/system/IO-TCP/selectors/$a/0] Connection established to [localhost:55653]
[DEBUG] [02/13/2018 11:43:10.291] [default-akka.actor.default-dispatcher-5] [default/Pool(shared->http://localhost:55653)] [0] </127.0.0.1:55669->localhost:55653> Received response: GET /search/category Empty -> 200 OK Strict(35 bytes)
[DEBUG] [02/13/2018 11:43:10.296] [default-akka.actor.default-dispatcher-8] [default/Pool(shared->http://localhost:55653)] [0] </127.0.0.1:55669->localhost:55653> Finished reading response entity for GET /search/category Empty -> 200 OK Strict(35 bytes)
[DEBUG] [02/13/2018 11:43:10.298] [default-akka.actor.default-dispatcher-5] [default/Pool(shared->http://localhost:55653)] [0] Loaded(1) -> Idle
[ScalaTest-run-running-MyLibraryClientPactSpec] INFO org.http4s.blaze.channel.ServerChannel - Closing NIO1 channel /127.0.0.1:55653 at Tue Feb 13 11:43:10 GMT 2018
[ScalaTest-run-running-MyLibraryClientPactSpec] INFO org.http4s.blaze.channel.nio1.NIO1SocketServerGroup - Closing NIO1SocketServerGroup
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.nio1.SelectorLoop - Shutting down SelectorLoop blaze-nio-fixed-selector-pool-0
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.nio1.SelectorLoop - Shutting down SelectorLoop blaze-nio-fixed-selector-pool-1
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.nio1.SelectorLoop - Shutting down SelectorLoop blaze-nio-fixed-selector-pool-2
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.nio1.SelectorLoop - Shutting down SelectorLoop blaze-nio-fixed-selector-pool-3
[blaze-nio1-acceptor] INFO org.http4s.blaze.channel.nio1.SelectorLoop - Shutting down SelectorLoop blaze-nio-fixed-selector-pool-4
[DEBUG] [02/13/2018 11:43:10.355] [default-akka.actor.default-dispatcher-3] [default/Pool(shared->http://localhost:55653)] [0] </127.0.0.1:55669->localhost:55653> connection was closed by peer while no requests were in flight
[DEBUG] [02/13/2018 11:43:10.360] [default-akka.actor.default-dispatcher-3] [default/Pool(shared->http://localhost:55653)] [0] Idle -> Unconnected
Process finished with exit code 0

我已经使用IntelliJ IDEA CE来执行测试，但是您可以直接使用这些命令来使用sbt：

• sbt test ：它执行扩展了 FunSpec和WordSpec的 所有测试（如在build.sbt定义）
• sbt pactTest ：它执行所有pacts测试

该测试验证了消费者协议，并 生成提供者 必须遵守 的契约/协议 。你可以找到它们，它们是遵循特定Pact结构的JSON文件。生成的应该是这样的： target/pacts

ScalaConsumer_myLibraryServer.json

{
 "provider" : {
   "name" : "myLibraryServer"
 },
 "consumer" : {
   "name" : "ScalaConsumer"
 },
 "interactions" : [
   {
     "request" : {
       "method" : "GET",
       "path" : "/search/category"
     },
     "description" : "Fetching categories",
     "response" : {
       "status" : 200,
       "headers" : {
         "Content-Type" : "application/json"
       },
       "body" : [
         {
           "name" : "Java"
         },
         {
           "name" : "DevOps"
         }
       ]
     },
     "providerState" : "Categories: [Java, DevOps]"
   }
 ]
}

正如你所看到的，这非常简单，两个参与者（提供者和消费者）的定义与可能的交互。

迄今为止已经很好好。但您可以添加更多的逻辑，更多的客户端，更多的契约，更多的服务等.Git仓库中的项目还包含一个小型服务，其中包含业务逻辑，计算类别的详细任务。这里是代码： CategoriesServiceSpec.scala

package com.fm.mylibrary.consumer.service
import com.fm.mylibrary.consumer.MyLibraryClient
import com.fm.mylibrary.model.Category
import org.scalamock.scalatest.MockFactory
import org.scalatest.{Matchers, WordSpec}
class CategoriesServiceSpec extends WordSpec with Matchers with MockFactory {
 private val mockMyLibraryClient = mock[MyLibraryClient]
 private val service = new CategoriesService(mockMyLibraryClient)
 "Count Categories" must {
   "return the number of all categories fetched form MyLibrary" in {
     val javaCategory = Category("Java")
     val devopsCategory = Category("DevOps")
     (mockMyLibraryClient.fetchCategories _).expects().returning(Some(List(javaCategory, devopsCategory)))
     val result = service.countCategories()
     result shouldBe 2
   }
   "return 0 in case of the fetch form MyLibrary fails" in {
     (mockMyLibraryClient.fetchCategories _).expects().returning(None)
     val result = service.countCategories()
     result shouldBe 0
   }
 }
}

CategoriesService.scala

package com.fm.mylibrary.consumer.service
import com.fm.mylibrary.consumer.MyLibraryClient
class CategoriesService(val myLibraryClient: MyLibraryClient) extends {
 def countCategories(): Int = myLibraryClient.fetchCategories() match {
   case None => 0
   case Some(categories) =>
     categories.size
 }
}

我没有使用任何依赖注入框架，因为我相信，如果微服务需要一个DI框架，那会使它变得非常庞大而复杂，但是如果你不像我这样想，可以随意使用它。我过去使用过 Google Guice ，看起来相当不错。

生产者（Provider）实现

一旦我们用契约文件定义了我们的消费者（ Consumer ），我们就可以转移到生产者并使用消费者产生的关联来实现它。

与往常一样，我们从测试开始。至于生产者，我们将有两种类型的测试，一种是验证协议，另一种是详细验证业务逻辑（单元测试）。服务器的实现通常比客户端要大得多，所以我认为最好从单元测试开始，一旦我们有了一个完整的应用程序，我们就可以创建测试来验证pact（或契约）。

另外，我总是建议采用增量方法（即使是小型项目），所以在这种情况下，我们可以构建一个服务器来公开一个API并返回两个类别的静态列表（如Pact文件中定义的），然后添加配置支持，数据库支持，迁移支持等。

在这里，我们将对我们的API进行单元测试：

CategoriesRoutesSpec.scala

package com.fm.mylibrary.producer
import com.fm.mylibrary.model.Category
import com.fm.mylibrary.model.JsonProtocol._
class CategoriesRoutesSpec extends BaseTestAppServer {
 "The service" should {
   "return an empty JSon array if there are no categories" in {
     Get("/search/category") ~> routes ~> check {
       responseAs[List[Category]] shouldBe List(Category("DevOps"), Category("Java"))
     }
   }
 }
}

以及具有所有测试依赖性的基本测试类BaseTestAppServer：

BaseTestAppServer.scala

package com.fm.mylibrary.producer
import akka.http.scaladsl.testkit.ScalatestRouteTest
import org.scalamock.scalatest.MockFactory
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpec}
import scala.concurrent.ExecutionContextExecutor
class BaseTestAppServer extends WordSpec
           with ScalatestRouteTest
           with Matchers
           with MockFactory
           with Routes
           with BeforeAndAfterAll {
 implicit val executionContext: ExecutionContextExecutor = system.dispatcher
}

该测试是使用Akka HTTP Route TestKit实现的，您可以在这里找到官方文档，它允许在这种格式的路由上构建测试：

REQUEST ~> ROUTE ~> check {
    ASSERTIONS 
}

BaseTestAppServer 的类包含基本的依赖 WordSpec ， ScalatestRouteTest ， Matchers ， MockFactory ， BeforeAndAfterAll 和定义应用程序的路由的性状： Routes

当然它不会编译也不会传递，因为还没有实现，所以让我们定义我们的路由：

Routes.scala

package com.fm.mylibrary.producer
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.Route
import akka.stream.Materializer
import com.fm.mylibrary.model.Category
import scala.concurrent.ExecutionContext
import spray.json._
import com.fm.mylibrary.model.JsonProtocol._
trait Routes {
 implicit val materializer: Materializer
 implicit val executionContext: ExecutionContext
 val searchRoutes: Route = {
   pathPrefix("search" / "category") {
     get {
       complete(
         List(Category("DevOps"), Category("Java")).toJson
       )
     }
   }
 }
 val routes: Route = searchRoutes
}

我为json编组/解组使用了 spray-json ，并且它需要定义用于转换的协议（或格式），您可以在代码 import com.fm.mylibrary.model.JsonProtocol._ 中看到此对象的导入：; 还需要导入其中 import spray.json._ 提供转换的所有功能; 在这种情况下，我正在使用 toJson 寻找它将要转换的特定对象的协议（或格式）的隐式定义。

JsonProtocol.scala

package com.fm.mylibrary.model
import spray.json._
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
object JsonProtocol extends SprayJsonSupport with DefaultJsonProtocol {
 implicit val categoryFormat = jsonFormat1(Category)
}

没有必要为对象定义转换器 List ， Array ， Options ，等等，因为它们是由 DefaultJsonProtocol中 的，spry-json提供。

还有其他类似的库，如Argonaut和JSON4S，可以按你想法评估所有这些库，并选择最适合您需求的库。

如果我们再次执行测试，我们现在应该得到一条绿线。再次，添加更多的测试，以涵盖每一个案例。在此之前，为了检查我们的服务是否符合消费者契约，我们必须完成定义Akka HTTP应用程序的基本服务：

MyLibraryAppServer.scala

package com.fm.mylibrary.producer.app
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.directives.DebuggingDirectives
import akka.stream.ActorMaterializer
import com.fm.mylibrary.producer.Routes
import scala.concurrent.ExecutionContextExecutor
import scala.util.{Failure, Success}
object MyLibraryAppServer extends App
         with Routes
         with DebuggingDirectives {
 implicit val actorSystem: ActorSystem = ActorSystem()
 implicit val materializer: ActorMaterializer = ActorMaterializer()
 implicit val executionContext: ExecutionContextExecutor = actorSystem.dispatcher
 val log = actorSystem.log
 def startApplication(): Unit = {
   Http().bindAndHandle(handler = logRequestResult("log")(routes), interface = "localhost", port = 9000).onComplete {
     case Success(b) => log.info(s"application is up and running at ${b.localAddress.getHostName}:${b.localAddress.getPort}")
     case Failure(e) => log.error(s"could not start application: {}", e.getMessage)
   }
 }
 def stopApplication(): Unit = {
   actorSystem.terminate()
 }
 startApplication()
}

这个类定义了两个方法，一个是启动我们的服务器所必需的，另一个是停止服务器的方法，它还定义了将在路由处理中使用的actor系统和执行上下文。

它扩展了提供主要方法的特征 scala.App ，所以你可以执行这个类，它将启动一个提供定义路由的http服务器。

但首先，让我们来检查一下协议是否被满足，我们可以很容易地用这样的测试类来验证它：

MyLibraryServerPactSpec.scala

package com.fm.mylibrary.producer.pact
import com.fm.mylibrary.producer.app.MyLibraryAppServer
import org.scalatest.{BeforeAndAfterAll, FunSpec, Matchers}
import com.itv.scalapact.ScalaPactVerify._
class MyLibraryServerPactSpec extends FunSpec with Matchers with BeforeAndAfterAll {
 override def beforeAll() {
   MyLibraryAppServer.main(Array())
 }
 override def afterAll() {
   MyLibraryAppServer.stopApplication()
 }
 describe("Verifying MyLibrary server") {
   it("should be able to respect the contract"){
     verifyPact
       .withPactSource(loadFromLocal("target/pacts"))
       .noSetupRequired
       .runVerificationAgainst("localhost", 9999)
   }
 }
}

它使用可以以像类似 forgePact 方式使用的对象 verifyPact ，Pact文件的来源 target/pacts 在我们的例子中定义（但可以是共享位置或 Pact Broker ），设置执行所需的数据或环境所需的最终代码所有交互，然后是服务器正在侦听请求的主机和端口。

因此，根据Consumer测试，我们希望scala-pact执行真正的HTTP调用，所以我们需要设置应用程序以处理此调用。我们可以通过多种方式做到这一点，我为我选择了安全和简单的解决方案，即在生产中启动服务器，调用之前执行测试 MyLibraryAppServer 的主要方法，并且之后关闭它。如果应用程序很简单，我们可以使用这种方法，如果不是这样，我们可以为这种测试实现特定的测试运行器，但我建议尽可能与生产案例类似。

执行测试，我们应该得到一个pass和一个这样的输出：

[DEBUG] [02/13/2018 16:45:09.053] [ScalaTest-run] [EventStream(akka://default)] logger log1-Logging$DefaultLogger started
[DEBUG] [02/13/2018 16:45:09.054] [ScalaTest-run] [EventStream(akka://default)] Default Loggers started
[DEBUG] [02/13/2018 16:45:09.110] [ScalaTest-run] [AkkaSSLConfig(akka://default)] Initializing AkkaSSLConfig extension...
[DEBUG] [02/13/2018 16:45:09.112] [ScalaTest-run] [AkkaSSLConfig(akka://default)] buildHostnameVerifier: created hostname verifier: com.typesafe.sslconfig.ssl.DefaultHostnameVerifier@1bb571c
[DEBUG] [02/13/2018 16:45:10.244] [default-akka.actor.default-dispatcher-3] [akka://default/system/IO-TCP/selectors/$a/0] Successfully bound to /127.0.0.1:9000
[INFO] [02/13/2018 16:45:10.256] [default-akka.actor.default-dispatcher-3] [akka.actor.ActorSystemImpl(default)] application is up and running at 127.0.0.1:9000
Attempting to use local pact files at: 'target/pacts'
Looking for pact files in: target/pacts
Found directory: C:\Dev\git-1.0.6\home\src-rnd\myLibrary-contracts\target\pacts
Loading pact file: ScalaConsumer_myLibraryServer.json
Verifying against 'localhost' on port '9000' with a timeout of 2 second(s).
--------------------
Attempting to run provider state: Categories: [Java, DevOps]
Provider state ran successfully
--------------------
[DEBUG] [02/13/2018 16:45:10.883] [default-akka.actor.default-dispatcher-4] [akka://default/system/IO-TCP/selectors/$a/0] New connection accepted
[DEBUG] [02/13/2018 16:45:11.146] [default-akka.actor.default-dispatcher-2] [akka.actor.ActorSystemImpl(default)] log: Response for
  Request : HttpRequest(HttpMethod(GET),http://localhost:9000/search/category,List(Host: localhost:9000, User-Agent: scala-pact/0.16.2, Timeout-Access: <function1>),HttpEntity.Strict(none/none,ByteString()),HttpProtocol(HTTP/1.1))
  Response: Complete(HttpResponse(200 OK,List(),HttpEntity.Strict(application/json,[{"name":"DevOps"},{"name":"Java"}]),HttpProtocol(HTTP/1.1)))
[http4s-blaze-client-1] INFO org.http4s.client.PoolManager - Shutting down connection pool: allocated=1 idleQueue.size=1 waitQueue.size=0
[DEBUG] [02/13/2018 16:45:11.262] [default-akka.actor.default-dispatcher-2] [akka://default/system/IO-TCP/selectors/$a/1] Closing connection due to IO error java.io.IOException: An existing connection was forcibly closed by the remote host
Results for pact between ScalaConsumer and myLibraryServer
 - [  OK  ] Fetching categories
[DEBUG] [02/13/2018 16:45:11.391] [default-akka.actor.default-dispatcher-9] [EventStream] shutting down: StandardOutLogger started
[DEBUG] [02/13/2018 16:45:11.391] [default-akka.actor.default-dispatcher-7] [akka://default/system/IO-TCP/selectors/$a/0] Monitored actor [Actor[akka://default/user/StreamSupervisor-0/$a#-487633161]] terminated
Process finished with exit code 0

如果你不能执行，请确保在其中包含协议文件。 target/pactsMyLibraryClientPactSpec

消费者协议似乎受到尊重，所以我们可以继续实现，添加外部配置文件，数据库支持和数据库迁移支持。

添加外部配置是很容易的，只需要在创建文件下，配置它所有的配置值，即： application.confsrc/main/resources

application.conf

akka {
 loglevel = DEBUG
}
http {
 interface = "0.0.0.0"
 port = 9000
}
database = {
 url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1"
 driver = org.h2.Driver
 connectionPool = disabled
 keepAliveConnection = true
 user = "sa"
 password = ""
}

然后，您可以创建一个处理它的特征，从而加载配置和相应的命名常量：

Config.scala

package com.fm.mylibrary.producer
import com.typesafe.config.ConfigFactory
trait Config {
 private val config = ConfigFactory.load()
 private val httpConfig = config.getConfig("http")
 private val databaseConfig = config.getConfig("database")
 val httpInterface: String = httpConfig.getString("interface")
 val httpPort: Int = httpConfig.getInt("port")
 val databaseUrl: String = databaseConfig.getString("url")
 val databaseUser: String = databaseConfig.getString("user")
 val databasePassword: String = databaseConfig.getString("password")
}

默认情况下， ConfigFactory.load() 从 src/main/resources/application.conf 该位置加载配置

我们也可以将测试的配置版本放在： src/test/resources

application.conf

akka {
 loglevel = DEBUG
}
http {
 interface = "localhost"
 port = 9999
}
database = {
 url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1"
 driver = org.h2.Driver
 connectionPool = disabled
 keepAliveConnection = true
 user = "sa"
 password = ""
}

在这种情况下没有太大的不同，因为我正在使用内存数据库。

在主类中使用它非常容易; 只需将其添加为类特征，并将静态值替换为相应的常量即可：

MyLibraryAppServer.scala

package com.fm.mylibrary.producer.app
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.directives.DebuggingDirectives
import akka.stream.ActorMaterializer
import com.fm.mylibrary.producer.{Config, Routes}
import scala.concurrent.ExecutionContextExecutor
import scala.util.{Failure, Success}
object MyLibraryAppServer extends App
         with Routes
         with Config
         with DebuggingDirectives {
 implicit val actorSystem: ActorSystem = ActorSystem()
 implicit val materializer: ActorMaterializer = ActorMaterializer()
 implicit val executionContext: ExecutionContextExecutor = actorSystem.dispatcher
 val log = actorSystem.log
 def startApplication(): Unit = {
   Http().bindAndHandle(handler = logRequestResult("log")(routes), interface = httpInterface, port = httpPort).onComplete {
     case Success(b) => log.info(s"application is up and running at ${b.localAddress.getHostName}:${b.localAddress.getPort}")
     case Failure(e) => log.error(s"could not start application: {}", e.getMessage)
   }
 }
 def stopApplication(): Unit = {
   actorSystem.terminate()
 }
 startApplication()
}

您也可以在Pact测试中使用该配置，以便使用正确的服务器地址：

MyLibraryServerPactSpec.scala

package com.fm.mylibrary.producer.pact
import com.fm.mylibrary.producer.Config
import com.fm.mylibrary.producer.app.MyLibraryAppServer
import org.scalatest.{BeforeAndAfterAll, FunSpec, Matchers}
import com.itv.scalapact.ScalaPactVerify._
class MyLibraryServerPactSpec extends FunSpec with Matchers with BeforeAndAfterAll with Config {
 override def beforeAll() {
   MyLibraryAppServer.main(Array())
 }
 override def afterAll() {
   MyLibraryAppServer.stopApplication()
 }
 describe("Verifying MyLibrary server") {
   it("should be able to respect the contract"){
     verifyPact
       .withPactSource(loadFromLocal("target/pacts"))
       .noSetupRequired
       .runVerificationAgainst(httpInterface, httpPort)
   }
 }
}

现在我们终于可以通过迁移来添加数据库支持。

首先，我们必须定义我们的实体（或表），在我们的例子中，我们只需要一个： Category

CategoryEntity.scala

package com.fm.mylibrary.producer.entity
import com.fm.mylibrary.model.Category
import slick.jdbc.H2Profile.api._
trait CategoryEntity {
 class Categories(tag: Tag) extends Table[Category](tag, "CATEGORY") {
   def name = column[String]("NAME", O.PrimaryKey)
   def * = name  <> (Category.apply, Category.unapply)
 }
 protected val categories = TableQuery[Categories]
}

这是一个标准的光滑表格定义; 你可以看到这个表只有一列也是主键，它和类的类别有关 Table[Category]

它可以从Category类中实例化，如定义：def * = name <> (Category.apply, Category.unapply)，确保模型类同时实现了apply和unapply，最简单的方法是定义模型类的案例类。

最后一条指令是定义TableQuery对象，该对象对于该表执行任何类型的查询都是必需的。让我们来定义我们的任何数据库交互的主要入口点，我已经实现了它可以被任何类需要数据库访问使用的特征：

DatabaseSupport.scala

package com.fm.mylibrary.producer.db
import slick.jdbc.H2Profile
import slick.jdbc.H2Profile.api._
trait DatabaseSupport {
 val db: H2Profile.backend.Database = Database.forConfig("database")
 def closeDB(): Unit = db.close
}

我们现在可以定义在类别表DAO上操作所必需的图层。我已经在 CategoryEntity的 相同的文件中创建了它，但是如果您想要使用不同的包，则可以将它移动到不同的文件中：

CategoryEntity.scala

package com.fm.mylibrary.producer.entity
import com.fm.mylibrary.model.Category
import com.fm.mylibrary.producer.db.DatabaseSupport
import slick.jdbc.H2Profile.api._
import scala.concurrent.Future
trait CategoryEntity {
 class Categories(tag: Tag) extends Table[Category](tag, "CATEGORY") {
   def name = column[String]("NAME", O.PrimaryKey)
   def * = name  <> (Category.apply, Category.unapply)
 }
 protected val categories = TableQuery[Categories]
}
class CategoryDAO extends CategoryEntity with DatabaseSupport {
 def insertOrUpdate(category: Category): Future[Int] =
       db.run(categories.insertOrUpdate(category))
 def findAll(): Future[Seq[Category]] =
       db.run(categories.result)
}

CategoryDAO 同时扩展 DatabaseSupport 和 CategoryEntity ，首先是要获得分类表查询的对象，第二个是要得到数据库实例用来执行查询。

我只实现了两种方法，对我们的测试来说已经足够了。正如您所看到的，我使用Slick提供的基本方法，并且由于实体 Categories 和模型 Category 相互关联，因此DAO可以直接返回模型而不显式转换。您可以在官方文档中找到更多关于如何在Slick中实现实体和DAO的示例和信息。

如果他们实现库提供的标准查询，我通常不会实现DAO测试，我没有看到测试外部库方法的任何一点，并且它们已经被路由测试覆盖了。但是，如果DAO实现了涉及多个表的复杂查询，我强烈建议对所有可能的案例进行单元测试。

为了现在开始我们的应用程序，需要一个带有分类表的数据库，并且我们可以手动完成，或者让机器为我们完成工作。所以我们可以实现一个数据库迁移，它能够在启动时应用任何必要的数据库更改来执行应用程序。

正如我们为数据库支持所做的那样，我们可以实现一个提供执行迁移功能的特性：

DatabaseMigrationSupport.scala

package com.fm.mylibrary.producer.db
import com.fm.mylibrary.producer.Config
import org.flywaydb.core.Flyway
trait DatabaseMigrationSupport extends Config {
 private val flyway = new Flyway()
 flyway.setDataSource(databaseUrl, databaseUser, databasePassword)
 def migrateDB(): Unit = {
   flyway.migrate()
 }
 def reloadSchema(): Unit = {
   flyway.clean()
   flyway.migrate()
 }
}

这暴露了两种方法，一种是增量迁移，一种是重新执行整个迁移。它使用特征来获取数据库连接信息。 Config

默认情况下，Flayway会在 src/main/resources/db/migration中 查找迁移的 sql 脚本文件，它需要具有特定名称格式的文件：

从官方迁移文档获取更多信息。

所以，我们的第一个迁移脚本是创建分类表：

V1__Create_Category.sql

CREATE TABLE category (
 name VARCHAR(255) NOT NULL PRIMARY KEY
);

我们可以在服务器启动时执行它：

MyLibraryAppServer.scala

package com.fm.mylibrary.producer.app
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.directives.DebuggingDirectives
import akka.stream.ActorMaterializer
import com.fm.mylibrary.producer.db.DatabaseMigrationSupport
import com.fm.mylibrary.producer.{Config, Routes}
import scala.concurrent.ExecutionContextExecutor
import scala.util.{Failure, Success}
object MyLibraryAppServer extends App
         with Routes
         with Config
         with DatabaseMigrationSupport
         with DebuggingDirectives {
 implicit val actorSystem: ActorSystem = ActorSystem()
 implicit val materializer: ActorMaterializer = ActorMaterializer()
 implicit val executionContext: ExecutionContextExecutor = actorSystem.dispatcher
 val log = actorSystem.log
 def startApplication(): Unit = {
   migrateDB()
   Http().bindAndHandle(handler = logRequestResult("log")(routes), interface = httpInterface, port = httpPort).onComplete {
     case Success(b) => log.info(s"application is up and running at ${b.localAddress.getHostName}:${b.localAddress.getPort}")
     case Failure(e) => log.error(s"could not start application: {}", e.getMessage)
   }
 }
 def stopApplication(): Unit = {
   actorSystem.terminate()
 }
 startApplication()
}

我们在HTTP绑定之前添加了 DatabaseMigrationSupport和migrateDB() 的调用。

最后一件事是将我们的新数据源与业务逻辑关联起来，改变路线以便从DB中检索类别：

Routes.scala

package com.fm.mylibrary.producer
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.Route
import akka.stream.Materializer
import com.fm.mylibrary.producer.entity.CategoryDAO
import scala.concurrent.ExecutionContext
import spray.json._
import com.fm.mylibrary.model.JsonProtocol._
trait Routes {
 implicit val materializer: Materializer
 implicit val executionContext: ExecutionContext
 private val categoryEntityDAO = new CategoryDAO()
 val searchRoutes: Route = {
   pathPrefix("search" / "category") {
     get {
       complete(
         categoryEntityDAO.findAll()
             .map(_.toJson)
       )
     }
   }
 }
 val routes: Route = searchRoutes
}

我们刚刚调用dao中的 findAll 方法替换了静态列表。

你可以看到dao在trait中被实例化，如果逻辑变得更复杂，我建议将它作为必需的参数（隐式或类属性）移动，以便从外部注入它们。在我们现在的情况下，没有必要，因为逻辑非常简单，在测试方面，我们使用的是内存数据库，所以没有必要对它进行模拟。

回到测试路径上，它会失败，因为没有数据，所以我们要添加它们。我们可以很容易地用一种方法的特征来实现，这个特征实现了一个方法，添加了几个类别:：

MockData.data

package com.fm.mylibrary.producer.db
import com.fm.mylibrary.model.Category
import com.fm.mylibrary.producer.entity.CategoryDAO
import scala.concurrent.{Await, ExecutionContext}
import scala.concurrent.duration.Duration
trait MockData {
 implicit val executionContext: ExecutionContext
 def addMockCategories(): Unit = {
   val categoryEntityDAO = new CategoryDAO()
   val setupFuture = for {
     c1 <- categoryEntityDAO.insertOrUpdate(Category("Java"))
     c2 <- categoryEntityDAO.insertOrUpdate(Category("DevOps"))
   } yield c1 + c2
   Await.result(setupFuture, Duration.Inf)
 }
}

将它添加进来，以便我们可以使用路由测试和Pact测试轻松验证应用程序： BaseAppServerTestAppMyLibraryAppServer

MyLibraryAppServer.scala

package com.fm.mylibrary.producer.app
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.directives.DebuggingDirectives
import akka.stream.ActorMaterializer
import com.fm.mylibrary.producer.db.{DatabaseMigrationSupport, MockData}
import com.fm.mylibrary.producer.{Config, Routes}
import scala.concurrent.ExecutionContextExecutor
import scala.util.{Failure, Success}
object MyLibraryAppServer extends App
         with Routes
         with Config
         with DatabaseMigrationSupport
         with MockData
         with DebuggingDirectives {
 implicit val actorSystem: ActorSystem = ActorSystem()
 implicit val materializer: ActorMaterializer = ActorMaterializer()
 implicit val executionContext: ExecutionContextExecutor = actorSystem.dispatcher
 val log = actorSystem.log
 def startApplication(): Unit = {
   migrateDB()
   addMockCategories()
   Http().bindAndHandle(handler = logRequestResult("log")(routes), interface = httpInterface, port = httpPort).onComplete {
     case Success(b) => log.info(s"application is up and running at ${b.localAddress.getHostName}:${b.localAddress.getPort}")
     case Failure(e) => log.error(s"could not start application: {}", e.getMessage)
   }
 }
 def stopApplication(): Unit = {
   actorSystem.terminate()
 }
 startApplication()
}

BaseTestAppServer.scala

package com.fm.mylibrary.producer
import akka.http.scaladsl.testkit.ScalatestRouteTest
import com.fm.mylibrary.producer.db.{DatabaseMigrationSupport, MockData}
import org.scalamock.scalatest.MockFactory
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpec}
import scala.concurrent.ExecutionContextExecutor
class BaseTestAppServer extends WordSpec
           with ScalatestRouteTest
           with Matchers
           with MockFactory
           with DatabaseMigrationSupport
           with MockData
           with Routes
           with BeforeAndAfterAll {
 implicit val executionContext: ExecutionContextExecutor = system.dispatcher
 override def beforeAll(): Unit = {
   migrateDB()
   addMockCategories()
 }
}

如果我们执行所有测试，我们应该没有问题; 你可以用 sbt test 命令来做到这一点

如果我们启动服务器，用 sbt run命令 ，并执行 GET /search/category ，我们应该得到我们的两个类别：

总结

消费者驱动的契约测试是一项非常棒的技术，可以节省很多时间和与集成测试相关的问题。

所有的实现都是*“以 契约 为中心”的，*所以它意味着我们强制首先考虑如何让消费者获得特定的服务，并且我们必须提供特定的服务，然后我们不需要设置基础设施来执行集成测试服务。

另一方面，Scala协议没有很好的文档记录，因此设置复杂测试会很有挑战性，而我发现的唯一方法是浏览它的 示例 和源代码。

我们已经看到了一个非常简单的例子，很少在真实环境中使用，但是希望您可以将它用作下一个微服务的起点。

更多关于CDC和Pact

我已经向你展示了Pact的最基本用法，对于一个真正的环境来说这可能是不够的，因为有许多团队，每个团队都与许多生产者和消费者进行*“并发”*工作，其中通信非常重要，以及自动化和用于解决它的工具。

在CDC和Pact的情况下，您必须自动执行契约处理（发布/验证），并将其与CI / CD（持续集成/持续交付）流程相链接，以便在没有相关生产商的情况下客户无法投入生产尊重他们的契约，如果违反了某些契约，任何生产者都不能生产。

所以，我强烈建议您将 Pact 的官方文档和介绍人 Pact Broker 带入您的CI / CD流程，它是一个提供以下功能的应用程序（来自官方 文档 ）：

• 通过独立部署您的服务并避免集成测试的瓶颈，您可以快速，放心地利用客户价值
• 解决了如何在消费者和提供者项目之间共享契约验证结果的问题
• 告诉您可以将应用程序的哪个版本安全地部署在一起，自动地将您的合同版本部署在一起
• 允许您确保多个消费者版本和提供者版本之间的向后兼容性（例如，在移动或多租户环境中）
• 提供保证为最新的应用程序的API文档
• 向您展示您的服务如何互动的真实例子
• 允许您可视化服务之间的关系

您可以随时提出任何问题，如果您需要建议，我将非常乐意提供帮助。

扩展阅读：https://www.cnblogs.com/jinjiangongzuoshi/p/7815243.html

问答

微服务架构：跨服务数据共享如何实现？

相关阅读

在微服务之间进行通信

服务集成时需避免的两个错误

基于 RabbitMQ 和 AMQP 进行消息传递

此文已由作者授权腾讯云+社区发布，原文链接：https://cloud.tencent.com/developer/article/1149167?fromSource=waitui

欢迎大家前往腾讯云+社区或关注云加社区微信公众号（QcloudCommunity），第一时间获取更多海量技术实践干货哦~