simpMessagingTemplate convertAndSendToUser许多等待线程阻塞其他功能

Question

我们在应用程序中使用Stomp，SpringBoot和WebSockets。服务器应用程序正在执行以下操作：1）生成要推送给用户的消息，2）接受WebSocket连接和3）将消息推送到ActiveMQ stomp代理。线程转储显示与simpMessagingTemplate convertAndSendToUser API调用关联的许多等待线程。

应用程序的两个实例在云中运行。此应用程序使用simpMessagingTemplate convertAndSendToUser API生成消息并推送到ActiveMQ stomp代理（单独运行）。

我们使用Gatling模拟用户WebSocket连接以进行负载测试。 Gatling在一个单独的实例上运行。该应用程序适用于2000个用户连接。一旦我们将用户增加到4000，我们就会看到消息生成线程停止。用户正在连接到相同的服务器，但没有任何问题。

如果我们评论simpMessagingTemplate convertAndSendToUser API调用，那么一切都很好（生成消息和新的WebSocket连接）。所以我们怀疑convertAndSendToUser API的问题。

Threaddump堆栈跟踪如下：

"ForkJoinPool-1-worker-440" #477 daemon prio=5 os_prio=0 tid=0x00007f0c541c2800 nid=0x2a47 sleeping[0x00007f08e6371000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
	at java.lang.Thread.sleep(Native Method)
	at reactor.util.concurrent.WaitStrategy$Sleeping.waitFor(WaitStrategy.java:319)
	at reactor.core.publisher.MonoProcessor.block(MonoProcessor.java:211)
	at reactor.core.publisher.MonoProcessor.block(MonoProcessor.java:176)
	at org.springframework.messaging.tcp.reactor.AbstractMonoToListenableFutureAdapter.get(AbstractMonoToListenableFutureAdapter.java:73)
	at org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler$SystemStompConnectionHandler.forward(StompBrokerRelayMessageHandler.java:980)
	at org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler.handleMessageInternal(StompBrokerRelayMessageHandler.java:549)
	at org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler.handleMessage(AbstractBrokerMessageHandler.java:234)
	at org.springframework.messaging.support.ExecutorSubscribableChannel$SendTask.run(ExecutorSubscribableChannel.java:138)
	at org.springframework.messaging.support.ExecutorSubscribableChannel.sendInternal(ExecutorSubscribableChannel.java:94)
	at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:119)
	at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:105)
	at org.springframework.messaging.simp.SimpMessagingTemplate.sendInternal(SimpMessagingTemplate.java:187)
	at org.springframework.messaging.simp.SimpMessagingTemplate.doSend(SimpMessagingTemplate.java:162)
	at org.springframework.messaging.simp.SimpMessagingTemplate.doSend(SimpMessagingTemplate.java:48)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)
	at org.springframework.messaging.simp.user.UserDestinationMessageHandler.handleMessage(UserDestinationMessageHandler.java:227)
	at org.springframework.messaging.support.ExecutorSubscribableChannel$SendTask.run(ExecutorSubscribableChannel.java:138)
	at org.springframework.messaging.support.ExecutorSubscribableChannel.sendInternal(ExecutorSubscribableChannel.java:94)
	at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:119)
	at org.springframework.messaging.simp.SimpMessagingTemplate.sendInternal(SimpMessagingTemplate.java:187)
	at org.springframework.messaging.simp.SimpMessagingTemplate.doSend(SimpMessagingTemplate.java:162)
	at org.springframework.messaging.simp.SimpMessagingTemplate.doSend(SimpMessagingTemplate.java:48)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)
	at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:150)
	at org.springframework.messaging.simp.SimpMessagingTemplate.convertAndSendToUser(SimpMessagingTemplate.java:229)
	at org.springframework.messaging.simp.SimpMessagingTemplate.convertAndSendToUser(SimpMessagingTemplate.java:218)
	at org.springframework.messaging.simp.SimpMessagingTemplate.convertAndSendToUser(SimpMessagingTemplate.java:204)
	at com.mypackage.PushMessageManager.lambda$sendMyMessage$2(PushMessageManager.java:77)
	at com.mypackage.PushMessageManager$$Lambda$923/1850582969.accept(Unknown Source)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
	at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
	at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinTask.doInvoke(ForkJoinTask.java:401)
	at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:734)
	at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:174)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
	at com.mypackage.PushMessageManager.sendMyMessage(PushMessageManager.java:74)
	at com.mypackage.PushMessageManager.lambda$processPushMessage$0(PushMessageManager.java:61)
	at com.mypackage.PushMessageManager$$Lambda$664/624459498.run(Unknown Source)
	at nl.talsmasoftware.context.functions.RunnableWithContext.run(RunnableWithContext.java:42)
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at nl.talsmasoftware.context.executors.ContextAwareExecutorService$1.call(ContextAwareExecutorService.java:59)
	at nl.talsmasoftware.context.delegation.RunnableAdapter.run(RunnableAdapter.java:44)
	at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

   Locked ownable synchronizers:
	- None

下面的步骤如下图所示：

1. Gatling JMS发布者以每分钟20000条消息将JMS消息推送到Active MQ代理。请注意，这些消息不仅适用于一个用户。它是基于WebSocket用户连接分发的。
2. 我们的应用程序有一个JMS侦听器来接收这些消息。我们运行说应用程序的2个实例，因此两个JMS侦听器处理此消息。
3. 一旦应用程序收到JMS消息，它就会从缓存中检查会话信息以识别连接的用户，并使用simpMessagingTemplate convertAndSendToUser API simpMessagingTemplate.convertAndSendToUser（sessionId，“/ queue / abc”，payload）将其推送到另一个ActiveMQ stomp代理。请注意，当用户首次连接到应用程序时，sessionId存储在分布式缓存中。所以这些是有效的会话ID。
4. 然后，ActiveMQ stomp代理将这些消息传播到单个用户的stomp队列。
5. Gatling WebSocket客户端（每个具有2000个用户连接）将通过WebSocket连接接收这些消息。
6. 客户端连接和订阅看起来像这样 stompClient.connect（{'username'：$（“＃userName”）。val（）}，function（frame）{setConnected（true）; subscription = stompClient.subscribe（'/ user / queue / abc'，function（message） ）{showData（JSON.parse（message.body））;}，headers = {'loginusername'：$（“＃userName”）。val（）}）;}）;

因此，每个用户只应收到针对他们的消息而不是所有消息。这就是我们在通过WebSocket连接时将用户连接到各个队列的原因，并且还使用convertAndSendToUser将消息推送到特定会话。后端JMS发布者确保以循环方式将消息发布给用户。

要回答有关识别瓶颈的问题，如果我们连接说2000用户一切正常。但是当我们添加更多用户时，我们发现应用程序的JMS侦听器无法监听后端Gatling JMS负载生成器发送的每分钟20000条消息。由此，ActiveMQ JMS队列深度增加。

为了确保瓶颈是convertAndSendToUser API，我们所做的就是注释了API调用。如果我们这样做，我们就能连接~13k WebSocket连接，后端JMS监听器也能够消耗每分钟消息的所有20000条消息。

希望这能澄清你的一些问题。 UPDATE代码片段用于显示simpMessagingTemplate.convertAndSendToUser API的异步调用，如下所示。这里RepositoryUtil.executor（）是我们自己的executor对象的包装器。

    public CompletableFuture<Void> processPushMessage(String userName, String payload) {
    return ContextAwareCompletableFuture.runAsync(() -> {
        sendABCMessage(payload, userName);
    }, RepositoryUtil.executor());
}

public void sendABCMessage(@Payload String payload, String username) {
    ArrayList<UserProfiles> userProfiles = (ArrayList<UserProfiles>) cacheService.getValue(username);
    if (Objects.nonNull(userProfiles) && userProfiles.size() > 0) {
      userProfiles.parallelStream()
          .filter(userProfiles1 -> ("/user/queue/abc".equalsIgnoreCase(userProfiles1.getSubscribeMapping()) && username.equals(userProfiles1.getUserName())))
          .forEach(userProfiles1 -> {              simpMessagingTemplate.convertAndSendToUser(userProfiles1.getSessionId(), "/queue/abc", payload);
          });
    } else {
      LOGGER.info("sendABCMessage userProfiles is null. Payload: {}", payload);
    }
}

Answer 1

我们可以通过移动到/ user / topic而不是/ user / queue来解决问题。我们现在能够处理来自后端和8k Web套接字用户连接的每分钟约35k的消息。

Answer 2

该应用程序适用于2000个用户连接，每分钟负载20,000条消息。一旦我们将用户增加到4000，我们就会看到消息生成线程停止。

如果将20,000条消息推送到ActiveMQ并且每条消息有1,000个订阅者，则意味着将20,000,000条消息（1,000 * 20,000）发布回WebSocket客户端。因此，请尝试确定流过的消息总量并了解瓶颈所在（服务器将消息转发到ActiveMQ，ActiveMQ处理消息或服务器向WebSocket客户端发布消息）。

对于20,000条消息，它们是从单个线程生成的，还是从大量不同的线程发送的，例如，处理来自WebSocket客户端或REST HTTP调用的消息的结果？如果是后者，可能是太多线程试图同时将消息转发给代理，您可能必须应用某种速率限制。

在一天结束时，您需要了解总体积，瓶颈所在，以及应用某些速率限制的位置。