请求的批处理问题及其解决方案（第1部分）

几乎所有现代软件产品都包含多种服务。 通常，较长的服务间通道响应时间会成为性能问题的根源。 解决此类问题的标准方法是将多个服务间请求打包到一个包中，这称为批处理。

如果使用批处理，则可能对性能或代码可理解性不满意。 对于呼叫者来说，此方法并不像您想象的那么容易。 对于不同的目的和情况，决策可能会有很大的不同。 在具体示例中，我将展示几种方法的利弊。

示范项目

为了清楚起见，请考虑我当前正在处理的应用程序中的一项服务的示例。


有文件批准服务。 有人创建一个文档并将其提交进行讨论，然后进行编辑，最终使文档保持一致。 协调服务本身对文件一无所知：它只是协调员的聊天，带有一些附加功能，我们在这里不予考虑。

因此，存在聊天室（对应于文档），每个聊天室中都有一组预定义的参与者。 与常规聊天一样，消息包含文本和文件，并且可以作为答复和转发：

data class ChatMessage (
   // nullable persist
   val id : Long ? = null ,
   /** */
   val author : UserReference ,
   /** */
   val message : String ,
   /** */
   // - JPA+ null,
   val files : List < FileReference > ? = null ,
   /** , */
   val replyTo : ChatMessage ? = null ,
   /** , */
   val forwardFrom : ChatMessage ? = null
)

指向文件和用户的链接是指向其他域的链接。 它像这样与我们一起生活：

typealias FileReference = Long
typealias UserReference = Long

用户数据存储在Keycloak中，并通过REST检索。 文件也是如此：文件和有关它们的元信息位于单独的文件存储服务中。

对这些服务的所有呼叫都是繁重的请求 。 这意味着传输这些请求的开销远远大于第三方服务处理这些请求所花费的时间。 在我们的测试台上，此类服务的典型呼叫时间为100毫秒，因此将来我们将使用这些数字。

我们需要制作一个简单的REST控制器，以接收包含所有必要信息的最后N条消息。 也就是说，我们认为在前端，消息模型几乎相同，因此我们需要发送所有数据。 前端模型之间的区别在于，文件和用户需要以略微解密的形式显示以使其链接：

/** */
data class ReferenceUI (
   /** url */
   val ref : String ,
   /** */
   val name : String
)
data class ChatMessageUI (
   val id : Long ,
   /** */
   val author : ReferenceUI ,
   /** */
   val message : String ,
   /** */
   val files : List < ReferenceUI >,
   /** , */
   val replyTo : ChatMessageUI ? = null ,
   /** , */
   val forwardFrom : ChatMessageUI ? = null
)

我们需要实现以下内容：

interface ChatRestApi {
   fun getLast ( n : Int ) : List < ChatMessageUI >
}

Postfix UI意味着前端的DTO模型，即我们必须通过REST提供的模型。

在这里我们似乎没有传递任何聊天标识符，甚至在ChatMessage / ChatMessageUI模型中也没有传递任何聊天标识符，这似乎令人惊讶。 我这样做是有意的，以免弄乱示例的代码（聊天是孤立的，因此我们可以假设我们只有一个）。


一个重要的限制：我们假设没有长久的响应或转发链。 也就是说，它们是，但是它们的长度不超过三个消息。 前端消息链必须完整传输。

要从外部服务接收数据，可以使用以下API：

interface ChatMessageRepository {
   fun findLast ( n : Int ) : List < ChatMessage >
}
data class FileHeadRemote (
   val id : FileReference ,
   val name : String
)
interface FileRemoteApi {
   fun getHeadById ( id : FileReference ) : FileHeadRemote
   fun getHeadsByIds ( id : Set < FileReference > ) : Set < FileHeadRemote >
   fun getHeadsByIds ( id : List < FileReference > ) : List < FileHeadRemote >
   fun getHeadsByChat () : List < FileHeadRemote >
}
data class UserRemote (
   val id : UserReference ,
   val name : String
)
interface UserRemoteApi {
   fun getUserById ( id : UserReference ) : UserRemote
   fun getUsersByIds ( id : Set < UserReference > ) : Set < UserRemote >
   fun getUsersByIds ( id : List < UserReference > ) : List < UserRemote >
}

可以看出，批处理最初是在外部服务中提供的，在两种情况下：通过Set（不保留元素的顺序，使用唯一键）和通过List（可能有重复项-保留了顺序）。

简单的实现

天真的实现

在大多数情况下，我们的REST控制器的第一个天真的实现将如下所示：

class ChatRestController (
   private val messageRepository : ChatMessageRepository ,
   private val userRepository : UserRemoteApi ,
   private val fileRepository : FileRemoteApi
) : ChatRestApi {
   override fun getLast ( n : Int ) =
     messageRepository . findLast ( n )
       . map { it . toFrontModel () }

   private fun ChatMessage . toFrontModel () : ChatMessageUI =
     ChatMessageUI (
       id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
       author = userRepository . getUserById ( author ) . toFrontReference () ,
       message = message ,
       files = files ?. let { files ->
         fileRepository . getHeadsByIds ( files )
           . map { it . toFrontReference () }
} ?: listOf () ,
       forwardFrom = forwardFrom ?. toFrontModel () ,
       replyTo = replyTo ?. toFrontModel ()
)
}

一切都非常清楚，这是一大优点。

我们使用批处理，并批量接收来自外部服务的数据。 但是性能如何？

对于每条消息，将对UserRemoteApi进行一次调用以获取作者字段中的数据，而对FileRemoteApi进行一次调用以接收所有附件。 似乎全部。 假定已获取ChatMessage的forwardFrom和ReplyTo字段，因此不需要额外的调用。 但是将它们转换为ChatMessageUI将导致递归，即呼叫计数的性能可以大大提高。 正如我们前面提到的，假设我们没有太多的嵌套，并且链条仅限于三封邮件。

结果，每个消息从两个到六个外部服务调用，到整个消息包的一个JPA调用。 通话总数从2 * N +1到6 * N +1。 实际单位是多少？ 假设您需要20个帖子才能呈现一个页面。 要获得它们，您需要4到10 s。 糟透了 我想满足500毫秒。 并且由于前端梦想实现无缝滚动，因此该端点的性能要求可以提高一倍。

优点：

1. 该代码简洁明了且具有自我记录功能（支持者的梦想）。
2. 代码很简单，因此几乎没有机会出手。
3. 批处理看起来并不陌生，并且有机地适合逻辑。
4. 逻辑更改将很容易进行，并且将是本地的。

减号：

由于包装很小，因此性能很差。

这种方法通常可以在简单的服务或原型中看到。 如果变更的速度很重要，则几乎不值得使系统复杂化。 同时，由于我们的服务非常简单，因此性能很差，因此这种方法的适用范围非常狭窄。

天真的并行处理

您可以开始并行处理所有消息-这将消除时间的线性增长，具体取决于消息的数量。 这不是一个特别好的方法，因为它会导致外部服务上的较大峰值负载。

实现并行处理非常简单：

override fun getLast ( n : Int ) =
   messageRepository . findLast ( n ) . parallelStream ()
     . map { it . toFrontModel () }
     . collect ( toList ())

使用并行消息处理，理想情况下，我们可以获得300-700 ms，这比单纯的实现要好得多，但仍然不够快。

使用这种方法，对userRepository和fileRepository的请求将同步执行，这不是很有效。 要解决此问题，您将不得不大量更改调用逻辑。 例如，通过CompletionStage（又名CompletableFuture）：

private fun ChatMessage . toFrontModel () : ChatMessageUI =
   CompletableFuture . supplyAsync {
     userRepository . getUserById ( author ) . toFrontReference ()
   } . thenCombine (
     files ?. let {
       CompletableFuture . supplyAsync {
         fileRepository . getHeadsByIds ( files ) . map { it . toFrontReference () }
}
} ?: CompletableFuture . completedFuture ( listOf ())
) { author , files ->
     ChatMessageUI (
       id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
       author = author ,
       message = message ,
       files = files ,
       forwardFrom = forwardFrom ?. toFrontModel () ,
       replyTo = replyTo ?. toFrontModel ()
)
   } . get () !!

可以看出，最初简单的映射代码变得不太清楚。 这是因为我们必须将外部服务调用与使用结果的位置分开。 这本身还不错。 但是调用的组合看起来并不优雅，类似于典型的响应式“面条”。

如果您使用协程，一切看起来会更加体面：

private fun ChatMessage . toFrontModel () : ChatMessageUI =
   join (
     { userRepository . getUserById ( author ) . toFrontReference () } ,
     { files ?. let { fileRepository . getHeadsByIds ( files )
       . map { it . toFrontReference () } } ?: listOf () }
   ) . let { ( author , files ) ->
     ChatMessageUI (
       id = id ?: throw IllegalStateException ( " $ this must be persisted" ) ,
       author = author ,
       message = message ,
       files = files ,
       forwardFrom = forwardFrom ?. toFrontModel () ,
       replyTo = replyTo ?. toFrontModel ()
)
   }

其中：

fun < A , B > join ( a : () -> A , b : () -> B ) =
   runBlocking ( IO ) {
     awaitAll ( async { a () } , async { b () } )
   } . let {
     it [ 0 ] as A to it [ 1 ] as B
   }

从理论上讲，使用这种并行处理，我们可以得到200-400毫秒，这已经接近我们的预期。

不幸的是，这种良好的并行化并没有发生，并且回报非常残酷：只有少数用户同时工作时，服务上会出现大量请求，这些请求仍然不会并行处理，因此我们将返回4s。

使用这种服务时，我的结果是处理20条消息的时间为1300-1700毫秒。 这比第一个实现要快，但是仍然不能解决问题。


优点：

1. 轻松实现并发消息处理。
2. 良好的可扩展性。

缺点：

1. 需要在对不同服务的并行处理请求中将数据的接收与其处理分开。
2. 第三方服务的负载增加。

可以看出，适用范围与幼稚方法大致相同。 如果由于他人的无情利用而想要多次提高服务性能，则使用并行查询方法才有意义。 在我们的示例中，生产率提高了2.5倍，但这显然是不够的。

快取

您可以为外部服务执行JPA样式的缓存，即在会话中存储接收到的对象，以免再次接收到它们（包括在批处理期间）。 您可以自己执行这些缓存，可以将Spring与其@Cacheable结合使用，并且始终可以手动使用像EhCache这样的现成缓存。

一般的问题将与以下事实有关：只有在命中时，缓存才具有良好的感觉。 在我们的案例中，极有可能在“作者”字段上命中（例如50％），并且根本没有文件命中。 这种方法将带来一些改进，但是性能不会发生根本变化（我们需要突破）。

会话（长）缓存需要复杂的失效逻辑。 通常，越晚解决会话间缓存的性能问题就越好。

优点：

1. 实现缓存而不更改代码。
2. 性能提高了数倍（在某些情况下）。

缺点：

1. 如果使用不当，可能会降低性能
2. 大量的内存开销，尤其是长缓存时。
3. 复杂的失效，错误将导致运行时出现难题。

通常，缓存仅用于快速修补设计问题。 这并不意味着不需要使用它们。 但是，始终值得谨慎对待它们，首先评估最终的性能提升，然后再做出决定。

在我们的示例中，缓存的性能将提高约25％。 同时，缓存也有很多缺点，因此我在这里不使用它们。

总结

因此，我们研究了使用批处理的服务的简单实现，以及一些加速它的简单方法。

所有这些方法的主要优点是简单，由此带来许多令人愉悦的结果。

这些方法的一个常见问题是性能下降，这主要是由于数据包大小所致。 因此，如果这些解决方案不适合您，则值得考虑采用更激进的方法。

您可以在两个主要领域中寻找解决方案：

• 数据异步处理（需要进行范式转换，因此不考虑本文）
• 在保持同步处理的同时扩大包装。

捆绑软件的扩大将大大减少外部调用的次数，同时使代码保持同步。 本文的下一部分将专门讨论该主题。