标签: spring-messaging

我在Tomcat 7上运行了一个Spring应用程序.添加WebSockets后,我开始在取消部署期间在日志中看到以下错误.

除了日志和增加取消部署时间(等待超时)之外,一切正常.

添加的WebSockets代码没有任何花哨的东西.只是一个@Controller,一个子类AbstractWebSocketMessageBrokerConfigurer和一个子类AbstractSecurityWebSocketMessageBrokerConfigurer.

怎么了？我不认为我错过了任何罐子......我可以org.springframework.context.support.DefaultLifecycleProcessor在罐子里看到spring-context-4.2.2-RELEASE我所拥有的课程WEB-INF/lib.那个极端的班级DefaultLifecycleProcessor$1也在那里.

14 Mar 2016 23:25:34,508  INFO  [context.support.DefaultLifecycleProcessor] Stopping beans in phase 2147483647
14 Mar 2016 23:25:34,508  DEBUG [context.support.DefaultLifecycleProcessor] Asking bean 'stompWebSocketHandlerMapping' of type [class org.springframework.web.socket.server.support.WebSocketHandlerMapping] to stop
14 Mar 2016 23:25:34,508  WARN  [context.support.DefaultLifecycleProcessor] Failed to stop bean 'stompWebSocketHandlerMapping'
java.lang.NoClassDefFoundError: org/springframework/context/support/DefaultLifecycleProcessor$1
    at org.springframework.context.support.DefaultLifecycleProcessor.doStop(DefaultLifecycleProcessor.java:229)
    at org.springframework.context.support.DefaultLifecycleProcessor.access$300(DefaultLifecycleProcessor.java:51)
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.stop(DefaultLifecycleProcessor.java:363)
    at org.springframework.context.support.DefaultLifecycleProcessor.stopBeans(DefaultLifecycleProcessor.java:202)
    at org.springframework.context.support.DefaultLifecycleProcessor.onClose(DefaultLifecycleProcessor.java:118)
    at org.springframework.context.support.AbstractApplicationContext.doClose(AbstractApplicationContext.java:969)
    at org.springframework.context.support.AbstractApplicationContext.close(AbstractApplicationContext.java:928)
    at org.springframework.web.servlet.FrameworkServlet.destroy(FrameworkServlet.java:828)
    at org.apache.catalina.core.StandardWrapper.unload(StandardWrapper.java:1481)
    at org.apache.catalina.core.StandardWrapper.stopInternal(StandardWrapper.java:1842) …

我spring-cloud-aws-messaging在 Spring Boot 项目中使用。
我在 AWS 中手动创建了 SQS 队列。

它的使用方式如下：

@SqsListener("${sqs.name.incoming}")
public void listen(String message) {
    ...
}

它工作正常。但是当我在 IDE 中停止我的应用程序时，或者当 Spring Boot 测试完成时，它会尝试停止队列。它无法阻止它并最终超时。它抛出这个异常：

2019-10-29 15:40:07.949  WARN 10378 --- [       Thread-2] s.c.a.m.l.SimpleMessageListenerContainer : An exception occurred while stopping queue 'my-awesome-queue-name'

java.util.concurrent.TimeoutException: null
    at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:204) ~[na:na]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer.waitForRunningQueuesToStop(SimpleMessageListenerContainer.java:161) ~[spring-cloud-aws-messaging-2.1.3.RELEASE.jar:2.1.3.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer.doStop(SimpleMessageListenerContainer.java:140) ~[spring-cloud-aws-messaging-2.1.3.RELEASE.jar:2.1.3.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.AbstractMessageListenerContainer.stop(AbstractMessageListenerContainer.java:351) ~[spring-cloud-aws-messaging-2.1.3.RELEASE.jar:2.1.3.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer.stop(SimpleMessageListenerContainer.java:45) ~[spring-cloud-aws-messaging-2.1.3.RELEASE.jar:2.1.3.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.AbstractMessageListenerContainer.stop(AbstractMessageListenerContainer.java:239) ~[spring-cloud-aws-messaging-2.1.3.RELEASE.jar:2.1.3.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer.stop(SimpleMessageListenerContainer.java:45) ~[spring-cloud-aws-messaging-2.1.3.RELEASE.jar:2.1.3.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.doStop(DefaultLifecycleProcessor.java:238) ~[spring-context-5.1.10.RELEASE.jar:5.1.10.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.access$300(DefaultLifecycleProcessor.java:53) ~[spring-context-5.1.10.RELEASE.jar:5.1.10.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.stop(DefaultLifecycleProcessor.java:377) ~[spring-context-5.1.10.RELEASE.jar:5.1.10.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.stopBeans(DefaultLifecycleProcessor.java:210) ~[spring-context-5.1.10.RELEASE.jar:5.1.10.RELEASE]
    at org.springframework.context.support.DefaultLifecycleProcessor.onClose(DefaultLifecycleProcessor.java:128) ~[spring-context-5.1.10.RELEASE.jar:5.1.10.RELEASE] …

我正在向客户发送价格(10000+)但是代码下面有循环导致客户等待计算的过程延迟.

PriceVisibleForCustomer =价格+ CustomerMargin

价格 - 每300毫秒更换一次 - 从中​​央商店发送,与客户实例无关

CustomerMargn - 由客户协议/分部/管理员决定等产生的一些加或减金额.在客户http会话期间没有变化,我可以将其保留在内存中

客户 - 他登录后参与了这个过程,他应该看到8种产品价格的快速变化.

也许我需要更多的技术？我有Spring 3/4,Java,Weblogic,我可以为此任务创建甚至单独的webapp,以提供计算价格.

我考虑过Java中的线程但是10000多个客户意味着太多的线程不是吗？如何更改此代码？也许我应该改变架构,但是如何？

/**
     * Sends prices to customer. This method is called very often (300ms) as prices are changing in real time.
     * Customer should see prices also each 300ms
     * @param productId - id of a product that prices will be calculated
     * @param productIdToPriceMap 
     * @param customerIdToMarginMap - this map is changed every time customer logs in or logs out
     */
    private …

问题
有没有办法全局处理MessageDeliveryExceptionSpring WebSocket 模块中由错误（通常权限不足）引起的Spring Messaging ？

用例
我已经通过 STOMP 实现了 Spring WebSockets 以支持我的 web 应用程序中的 ws 连接。为了保护 websocket 端点，我创建了授权用户在 STOMP CONNECT 时间启动 STOMP 会话的拦截器（如 Spring 文档中 22.4.11 部分的建议）：

@Component
public class StompMessagingInterceptor extends ChannelInterceptorAdapter {

    // Some code not important to the problem

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor headerAccessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);

        switch (headerAccessor.getCommand()) {
            // Authenticate STOMP session on CONNECT using jwt token passed as a STOMP login header - it's …

我们在我们的项目中使用 STOMP 代理中继（外部代理 - ActiveMQ 5.13.2）见 https://docs.spring.io/spring/docs/current/spring-framework-reference/web.html#websocket-stomp-handle-代理中继

我们使用以下堆栈：

org.springframework:spring-jms:jar:5.1.8.RELEASE
org.springframework:spring-messaging:jar:5.1.8.RELEASE
io.projectreactor:reactor-core:jar:3.2.8.RELEASE
io.projectreactor.netty:reactor-netty:jar:0.8.6.RELEASE
io.netty:netty-all:jar:4.1.34.Final

有时（假设每两周一次）我们可以在 tomcat catalina.out 中观察到日志错误

2019-08-21 13:38:58,891 [tcp-client-scheduler-5] ERROR com.*.websocket.stomp.SimpMessagingSender  - BrokerAvailabilityEvent[available=false, StompBrokerRelay[ReactorNettyTcpClient[reactor.netty.tcp.TcpClientDoOn@219abb46]]]
2019-08-21 13:38:58,965 [tcp-client-scheduler-1] ERROR org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler  - Transport failure: java.lang.IllegalStateException: No TcpConnection available

该错误之后STOMP通信中断（system连接 - 单个TCP连接不可用）

当我们从以下位置更新堆栈时，似乎一切都开始了：

org.springframework:spring-jms:jar:5.0.8.RELEASE
org.springframework:spring-messaging:jar:5.0.8.RELEASE
io.projectreactor:reactor-core:jar:3.1.8.RELEASE
io.projectreactor.netty:reactor-netty:jar:0.7.8.RELEASE
io.netty:netty-all:jar:4.1.25.Final

ActiveMQ 版本未更改

spring中有一个bug，system连接丢失时自动重新连接失败，请参见：https : //github.com/spring-projects/spring-framework/issues/22080

现在 3 个问题：

1. 如何使这个问题更具重现性？
2. 如何修复这种重新连接行为？:)
3. 如何防止失去这个连接？:)

编辑 23.09.2019

错误发生后，端口 61613（STOMP）的 TCP 堆栈如下（请注意 CLOSE_WAIT 状态）：

netstat -an | grep 61613
tcp6       0      0 :::61613                :::* …

这是一种“什么是@Service注解？” 问题，但采用另一种方法。因为，我不确定这里发生了什么：

我有一个控制器类：

@Controller
public class GreetingController {
    @Autowired
    SomeBean someBean;

    @MessageMapping("/msg")
    public String msg() {
        someBean.handleMsg();
        return "";
    }
}

我尝试从内部someBean.handleMsg向目的地发送响应。像这样的事情：

public class SomeBean {
    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    public handleMsg() {
        messagingTemplate.convertAndSend("/topic/someTopic", "Response");
    }
}

有两个版本的配置。

1. SomeBean 在.xml中配置：

喜欢：

< bean id="someBean" class="package.SomeBean"></bean>

2. SomeBean 被注释为服务（在第一个中它没有）：

喜欢：

@Service
public class SomeBean{...}

• 请注意，在这两种情况下，不存在任何有关注入等问题。在这两种情况下，客户端都会成功订阅、发送消息并处理消息。

唯一的 区别是：

• 当SomeBean有@Service注解时，成功响应客户端，但当没有注解时，客户端收不到响应消息，虽然没有任何异常。

这是问题：

• 在这种情况下@Service 实际上做了什么？有人可以解释一下这里发生了什么吗？

我尝试通过spring-cloud-stream使用测试消息发送和接收MessageCollector。

当我对流云使用本机 java 序列化时一切正常，但是当我将序列化更改为 json 时，MessageCollector 返回带有string有效负载而不是SomeObject有效负载的GenericMessage 。

配置是

  cloud.stream.default.contentType=application/json 

测试用例：

    Message outMsg = new GenericMessage<SomeObject>(new SomeObject(1));
    someChannel.send(outMsg);
    GenericMessage<SomeObject> inMsg = (GenericMessage<SomeObject>) messageCollector.forChannel(someChannel).poll();

    Assert.assertTrue(inMsg.getPayload() instanceof SomeObject);

因此，断言是错误的。inMsg包含字符串负载（字符串包含 SomeObject 的有效 json 表示）。

我的问题是：我如何GenericMessage从 SomeObject接收有效载荷MessageCollector？

生产环境工作正常，无需显式映射到SomeObject.

目前，我正在尝试使用 webflux 将 STOMP 与 websockets 一起使用。为了向 STOMP 主题发送消息，我需要使用SimpMessagingTemplate，这是我添加时由 spring boot 自动配置提供的@EnableWebSocketMessageBroker

但问题是，@EnableWebSocketMessageBroker间接地希望我spring-mvc在类路径中有库

@EnableWebSocketMessageBroker @ImportsDelegatingWebSocketMessageBrokerConfiguration扩展WebSocketMessageBrokerConfigurationSupport&WebSocketMessageBrokerConfigurationSupport#stompWebSocketHandlerMapping方法期望HandlerMapping返回该类

我的问题是

1. 如何在没有 webmvc 的情况下将 STOMP 与 webflux 集成
2. 为什么自动配置迫使我们在类路径中有 mvc（并且可能与 webflux 冲突）

我在通过 TCP 连接到 Spring Boot RSocket 应用程序时遇到问题。使用 RSocketRequester 时的客户端工作正常，但是当我尝试使用 RSocketFactory 客户端连接时，它不断出错。代码如下。

        RSocket rSocket = this.client = RSocketFactory
            .connect()
            .mimeType(WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.toString(), MediaType.APPLICATION_JSON_VALUE)
            .frameDecoder(PayloadDecoder.ZERO_COPY)
            .transport(TcpClientTransport.create("localhost", 7000))
            .start()
            .block();


Flux<Payload> s = rSocket.requestStream(DefaultPayload.create("1234", "socket"));
    s.subscribe();

这给出了如下错误：

java.lang.IndexOutOfBoundsException: readerIndex(1) + length(115) exceeds writerIndex(6): AbstractPooledDerivedByteBuf$PooledNonRetainedSlicedByteBuf(ridx: 1, widx: 6, cap: 6/6, unwrapped: PooledUnsafeDirectByteBuf(ridx: 27, widx: 27, cap: 1024))
at io.netty.buffer.AbstractByteBuf.checkReadableBytes0(AbstractByteBuf.java:1477)
at io.netty.buffer.AbstractByteBuf.checkReadableBytes(AbstractByteBuf.java:1463)
at io.netty.buffer.AbstractByteBuf.readSlice(AbstractByteBuf.java:880)
at io.rsocket.metadata.TaggingMetadata$1.next(TaggingMetadata.java:47)
at io.rsocket.metadata.TaggingMetadata$1.next(TaggingMetadata.java:37)
at org.springframework.messaging.rsocket.DefaultMetadataExtractor.extractEntry(DefaultMetadataExtractor.java:136)
at org.springframework.messaging.rsocket.DefaultMetadataExtractor.extract(DefaultMetadataExtractor.java:119)
at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.createHeaders(MessagingRSocket.java:195)
at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.handleAndReply(MessagingRSocket.java:167)
at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.requestStream(MessagingRSocket.java:127)
at io.rsocket.RSocketResponder.requestStream(RSocketResponder.java:207)
at io.rsocket.RSocketResponder.handleFrame(RSocketResponder.java:310)
at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
at …

我用教程创建了一个基本的网络套接字。

这是一个配置：

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic");
        config.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
         registry.addEndpoint("/chat");
         registry.addEndpoint("/chat").withSockJS();
    }
}

这是消息处理控制器：

@MessageMapping("/chat")
@SendTo("/topic/messages")
public OutputMessage send(Message message) throws Exception {
    return new OutputMessage("Hello World!");
}

一切正常，但根据我的调查，默认情况下 WebSockets 看起来有一个应用程序范围（通过连接到通道，我可以看到来自所有用户的所有调用）。

我想要做的是只能看到来自当前用户会话或当前视图的调用。关于如何应用这些配置的任何想法？