小贝子编程

simpMessagingTemplate convertAndSendToUser有很多等待线程阻塞了其他功能



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

该应用程序的两个实例正在云中运行。该应用程序使用simpMessagingTemplate convertAndSendToUser API生成消息并推送到ActiveMQ stomp broker(单独运行)。

我们使用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侦听器来接收这些消息。我们正在运行应用程序的两个实例，因此两个JMS侦听器将处理此消息
  3. 一旦应用程序接收到JMS消息，它将从缓存中检查会话信息以识别连接的用户，并使用simpMessagingTemplate convertAndSendToUser API simpMessagingTemplate.covertAndSendToUser(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)；subscribe=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.covertAndSendToUser 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);
}
}

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

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

如果向ActiveMQ推送20000条消息，并且每条消息都有1000个订阅者，则意味着20000000条消息(1000*20000)将发布回WebSocket客户端。因此，请尝试确定流经的消息的总量，并了解瓶颈在哪里(服务器将消息转发到ActiveMQ、ActiveMQ处理消息或服务器将消息发布到WebSocket客户端)。

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

在一天结束时，您需要了解整体流量、瓶颈在哪里以及在哪里应用一些速率限制。

相关内容

  • 没有找到相关文章

最新更新


热门标签：

javascript python java c# php android html jquery c++ css ios sql mysql arrays asp.net json python-3.x ruby-on-rails .net sql-server django objective-c excel regex ruby linux ajax iphone xml vba spring asp.net-mvc database wordpress string postgresql wpf windows xcode bash git oracle list vb.net multithreading eclipse algorithm macos powershell visual-studio image forms numpy scala function api selenium