wso2esb + websocket - ConcurrentModificationException error

Question

Я использую wso2esb + wso2mb + websockets для преобразования сообщений JMS из wso2mb в websockets. Во время тестирования производительности (я пытался отправить 7К сообщения от wso2mb к WebSocket) я получил следующее сообщение: errror

java.util.ConcurrentModificationException 
    at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:901) 
    at java.util.ArrayList$Itr.next(ArrayList.java:851) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.management.WebsocketSubscriberPathManager.broadcastOnSubscriberPath(WebsocketSubscriberPathManager.java:98) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketResponseSender.handleSendBack(InboundWebsocketResponseSender.java:117) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketResponseSender.sendBack(InboundWebsocketResponseSender.java:86) 
    at org.apache.synapse.core.axis2.Axis2Sender.sendBack(Axis2Sender.java:214) 
    at org.apache.synapse.mediators.builtin.RespondMediator.mediate(RespondMediator.java:35) 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:97) 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:59) 
    at org.apache.synapse.config.xml.AnonymousListMediator.mediate(AnonymousListMediator.java:37) 
    at org.apache.synapse.config.xml.SwitchCase.mediate(SwitchCase.java:69) 
    at org.apache.synapse.mediators.filters.SwitchMediator.mediate(SwitchMediator.java:119) 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:97) 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:59) 
    at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:158) 
    at org.apache.synapse.core.axis2.Axis2SynapseEnvironment.injectMessage(Axis2SynapseEnvironment.java:993) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.injectToSequence(InboundWebsocketSourceHandler.java:461) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.handleWebsocketPassthroughTextFrame(InboundWebsocketSourceHandler.java:346) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.handleWebSocketFrame(InboundWebsocketSourceHandler.java:242) 
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.channelRead(InboundWebsocketSourceHandler.java:132) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308) 
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294) 
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308) 
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294) 
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:244) 
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308) 
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294) 
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846) 
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131) 
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511) 
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468) 
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382) 
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354) 
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:110) 
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137) 
    at java.lang.Thread.run(Thread.java:745) 

Знаете ли вы, что может быть причиной? Сообщите мне, если вы хотите увидеть мой проект wso2esb.

Большое спасибо!

Answer 1

Когда новые клиенты подключены, они добавляются к карте. Если вы транслируете во время теста нагрузки, та же карта используется для добавления подключений и для вещания. Это закончится тем, что выступит с одновременным изменением. Это известное поведение, и мы рекомендуем прекратить трансляцию во время выполнения тестов производительности . Вместо этого напишите только один канал во время тестов производительности, и это решит проблему. Вы найдете код в [1] для дальнейшего анализа.

[1] https://github.com/wso2/carbon-mediation/blob/master/components/inbound-endpoints/org.wso2.carbon.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/websocket/management/WebsocketSubscriberPathManager.java#L98