想象一下，一根消防水龙带每天喷出数万亿加仑的水，你的一部分工作就是承受从中冒出的每一滴水。这就是可视化Apache Kafka®的消息吞吐量的感觉。

在Confluent，我们希望帮助开发人员了解如何考虑事件流以及它可以创造的机会。教育人们了解事件流是一项艰巨的任务。阅读这篇文章将让您深入了解我们如何构建一个高性能的UI来解决这个问题。

传统上，对分布式事件流平台中的数据流的理解是通过聚合数据的图表来完成的。然而，处理所有这些消息在计算上都是昂贵的，为了全面了解趋势而降低数据粒度是使数据和系统中发生的事情更容易理解的一种方法。随着可能有数百万条消息流经Kafka，如此高的消息吞吐量成为客户端读取数据和用户与系统交互的瓶颈。可能有成千上万台机器在翻阅数据，但当有人试图透过观察镜窥视时，用肉眼看是太快了。

如果有一种方法可以以最快的速度查看每条消息，而不必减慢速度呢？如果需要显示粒度呢？如果你想看到卡夫卡的每一条信息呢？好消息：这是可能的，你甚至可以在网络浏览器中完成。

在Kafka中为UI分页

一个典型的面试问题是：“你如何以高效的方式显示大量数据？“大多数人（至少在前端）通常会先想到分页。分页的实现可能如下：

在100项列表中，一次请求10项，直到达到100项为止。所以你会做9个请求，要求1-10，11-20，等等，直到100个。

在Kafka的例子中，连续请求之间可能有100万条消息，因此用户永远看不到“最新”消息，只能看到浏览器请求的范围。此外，与卡夫卡有关的分页还有一个根本问题。跨分区的消息排序是不确定的，因此在UI中显示的内容（从1到100的线性序列）不会表示数据，因为它是在Kafka中布局的。

浏览器还必须解析它接收到的数据并呈现它。根据负载大小或吞吐量，UI可能无法处理负载。

考虑到这种方法在显示事件流数据时的不精确性，需要有更好的解决方案。由于Kafka是一个事件流平台，因此处理此类问题的常见选项是长轮询。长轮询是一种拥有长寿命HTTP连接的方法。这给了你每个人都想要的实时性，但这是要付出代价的。

长期存在的HTTP请求有很大的请求开销，而且由于请求经常发出，因此速度成为一个问题。为了克服巨大的请求开销，与kafka接口的kafka rest包公开了rest端点，并将发送的消息以可配置的1秒间隔批处理。因为它是可配置的，所以有可能下载一些不可读的东西，或者每15分钟只发送一次新数据，这两种方式都会带来痛苦的用户体验。

我们需要的是给我们一个数据流，但很少有请求开销。

更好，更快，更强然后一切都会破碎

WebSockets不仅要求最小的请求开销，而且提供高吞吐量，从而解决了消息事件的性能问题。由于此方法不进行批处理，因此消息以实时处理速率到达它，并且由客户机处理执行情况。WebSocket不必一次处理一秒钟，而是允许我们在消息到达时就处理它们。我们接收的不是要解析的大块数据，而是可以处理得更快的小块数据。

在未来的迭代中，可以添加针对WebSocket暂停、重播或停止的功能。这为可以构建的应用程序类型提供了更大的灵活性，例如能够告诉端点每秒要使用多少消息，或者在不需要知道流中的数据的情况下重新启动数据流。

我们第一次尝试WebSockets时遇到了一些性能问题。接受来自WebSocket的消息、对其进行解析并将其发送到本地状态存储非常昂贵。我们只能解析、更新和呈现10条消息/秒，而浏览器不会变得无响应，消耗大量系统资源，最终不得不通过浏览器的任务管理器杀死它。

为了加快速度，所有其他消息都被留作字符串并放入缓冲区。当用户请求更新时，我们会切换缓冲列表，解析它，呈现它，然后重新填充缓冲区。这将浏览器锁定的程度降到了半合理的水平，但在大约每秒100条消息的情况下，浏览器将出现灾难性故障。

因此，我们开始对实现进行性能测试，以了解发生了什么。

开销太大

合流使用的本地状态商店（ local state store ）是Redux。简而言之，Redux维护一个大型JSON对象，并强制希望工程师将其中的数据变异为特定契约。这个契约有一些开销，但是使用单一方式与应用程序中的存储数据交互的好处远远超过了开销的负面影响。

在查看了WebSocket响应之后，很快就会发现Redux中触发操作的开销对于我们的应用程序来说是非常昂贵的。从测试中，执行一次解析、更新和呈现大约需要80毫秒。

根据单个操作的请求时间，~20条消息/秒是浏览器开始显示性能下降之前的上限，并且根据测试运行，不可恢复的浏览器锁定发生在~200条消息/秒左右。

这个测试的开销表明Redux的使用还有很大的改进空间。我们可以尽量把它移走，但仍然会有一些开销。如果我们把Redux状态换成react状态，我们可能会看到更好的性能。

当这个特性的Redux部分转换为react组件时，大多数调用的JavaScript执行量减少了40%。

为了更好地了解这是如何起作用的，我们采用了3秒的较大样本量，看到JavaScript执行量减少了26%。此外，浏览器能够通过所有的请求（空闲时间是这个指标）。

通过使用shouldComponentUpdate每六条消息对组件进行一次批量更新，进一步优化了这一点，如预期的那样，这会产生更高的性能结果。

批处理法的执行率进一步下降了66.9%。这意味着从Redux到这个小批处理，我们在Redux与批处理组件相同的3秒内将执行量减少了90%。

虽然这些改进是一个显著的性能提升，但是在锁定之前，UI吞吐量仅每秒增加大约90条消息，相对于可以通过主题的数千条消息/秒来说，这是很小的。这突出了在Kafka中显示消息的实际问题：随着消息吞吐量的增加，需要有一种不降低UI的方式来显示这些消息。

即使每个请求需要1毫秒才能完成，我们仍然会在UI上以每秒1000条消息的速度锁定处理。我们需要一种方法来保持UI的活跃性，同时最大化吞吐量。这是通过使用web工作者实现的。

更好，更快，更强，然后一切正常

web worker是一种将任务卸载到浏览器中不同线程上的方法。默认情况下，JavaScript只获取一个线程，即主UI线程。很少有网页需要超过一个线程，因为它们不做那么多工作。到目前为止所讨论的所有性能测试都是在主UI线程上进行的，因此当单个线程被淹没时就会发生锁定。

由于WebSockets具有处理大量数据的潜力，因此我们需要一种方法来保持UI响应性，同时从中获取尽可能多的数据。如果WebSocket留在主UI线程上，则保证会锁定浏览器，变得无响应并崩溃。

根据当前的修改，我们将浏览器锁定速度从每秒20条消息提高到每秒200条消息。最终系统中的某个地方会出现锁定，但是如果我们能够控制锁定发生的位置，那么如果我们使用workers，我们就可以从中恢复。

工作人员通过来回传递消息和广播到位于浏览器中不同沙盒中的事件侦听器进行操作。您可以将浏览器选项卡视为单个工作线程。从现有选项卡创建新选项卡时，它将创建一个与原始选项卡无关的完全独立的选项卡。为了从一个地方到另一个地方获取数据，可以将信息从一个地方复制/粘贴到另一个地方。这与web工作者的概念相同，只是在同一个浏览器选项卡中以编程方式发生。

不过，有一个问题。传递消息是昂贵的，因此实现web工作者必须小心。简单地来回传递单个消息不会提高性能。

因为在工作者和家长之间发送单个消息没有性能值，所以必须批处理。这类似于通过kafka rest进行的长轮询，但有一些关键的区别。首先，我们不使用HTTP节省了很多开销。一旦连接，WebSockets就使用TCP进行通信，并且由于UI控制轮询间隔，此方法为流式事件提供了最大的灵活性。通过定期对请求进行批处理，以从web工作者那里获得最大的性能增益，我们可以确保UI上的吞吐量显著提高。

从之前的测试中，请求将浏览器锁定在大约每秒200条消息。每当发生套接字事件时，我们都会看到大量的脚本执行。当我们看一看web worker内部执行的WebSocket时，我们注意到执行是完全不同的。为了举例说明，样本测试需要很长时间才能更好地显示正在发生的事情。

在14秒的测试中，web工作者能够解析所有事件并在150毫秒内做出响应。进一步观察web工作者，我们可以看到对消息的微秒响应。以前，WebSocket事件是为了获得针对主UI线程的计算时间而进行的。滚动事件、呈现和垃圾回收不再影响性能。web工作进程中发生的唯一执行是WebSocket事件。

当所有1200条消息都被发送回主UI线程时，渲染性能与预期一样快，每条消息的渲染时间为12毫秒。

我们现在可以看到在Worker内安装WebSocket的巨大好处。我们能够处理更大数量的数据，同时保持相当低的渲染时间。而且，正如我们可以通过工作进程中的空闲时间来判断的那样，我们可以处理更多的消息，甚至超出了测试显示的范围。

从消防水龙带发出的咯咯声

我们从一个WebSocket消息事件开始，这个事件花费了80ms，除了没有锁定UI之外，还将渲染时间减少到了12ms。相差147%！

由于所有空闲时间都在套接字中，对于这样的方法，似乎有一个很高的上限。然而，一旦我们达到足够高的WebSocket吞吐量，就应该在工作进程内部发生锁定。在这一点上，UI仍然可以使用，但是web工作者可能会导致问题。

为了最大化这一点，我们发现在WebSocket plus worker方法中每秒抛出5000条消息会导致290–300%的超高CPU利用率，并锁定web worker。由于该工作程序已被锁定执行，因此除了杀死违规的web工作程序外，没有其他办法。杀死Woker的唯一方法就是关闭标签。即使所有的事情都出了问题，风扇也像电脑要起飞一样炽热，用户界面仍然反应灵敏，但是用户电脑的损坏和使用是相当不受欢迎的。

对于这个问题，一个可能的，尽管有争议的解决方案是开始从socket事件中删除消息。虽然对于试图查看通过主题/流的所有消息的人来说，这在事实上并不准确，但这将防止锁定，允许我们在需要时优雅地关闭web套接字，并提醒用户由于吞吐量而处于降级状态。

以下是测试结果：

主UI线程是黄色的，工作线程是灰色的。

我们可以看到Worker花费了大量的时间和CPU使用量。主线程上的大部分大块来自页面上的其他XHR请求。解析消息的部分表明，大约需要250毫秒才能呈现大约300条消息。消息数量减少是由于节流阀。考虑到线性性能，每批处理7500条消息，当UI试图呈现这么大的负载时，它将锁定。为了达到这个吞吐量，Kafka需要每秒处理125000条消息。

下一步

如果您认为处理来自Kafka的200%以上的消息的UI实际上更好，请记住，它仍然可以远远超出当前可以处理的范围。如果将特定流的吞吐量考虑在内，我们将能够动态地限制和/或减少/增加批处理，或向服务器发送暂停事件，以便允许工作进程赶上。对于吞吐量较慢的主题，不需要限制。对于每秒数百万条消息或对于大量消息，可能需要一个大的限制和可能增加的呈现延迟。

另一个值得关注的是Redux的表现。我们的Redux使用中可能存在一些次优的模式，可以查看和修复这些模式，从而加快整个应用程序的速度。

我们还可以通过增加Worker的数量来分担负载，从而获得实质性的性能提升。目前，我们降级到一个端点进行消息传递。如果该端点被拆分，则可以扩大工作线程和套接字的数量，并使它们中的每一个都响应主线程。然后，我们可以利用React的异步呈现来获得更高的吞吐量，同时仍然保持响应性。

我们为什么要这么做？首先也是最重要的是，由于性能的原因，消息浏览器的第一个版本非常有限。虽然它像大多数现有的消息浏览器一样，只能在有限的消息数量范围内工作，但它不能很好地表示Kafka的功能。我们已经获得了如此多的性能，现在我们能够让用户更好地直观地了解卡夫卡和卡夫卡流中发生的事情，以及过滤、排序和搜索处理过的消息的能力，这在以前数量有限的情况下是不可能的。

 

原文：https://www.confluent.io/blog/consuming-messages-out-of-apache-kafka-in-a-browser/

本文：http://jiagoushi.pro/node/990

讨论：请加入知识星球或者微信圈子【首席架构师圈】