2

我在我的 Spring Boot 应用程序中为 websocket 配置外部代理（amazonMQ）中继。

在启动配置似乎很好，继电器按照下面的日志连接。我用错误的 URL 进行了测试，我显然得到了 UnknownHostException，所以我认为我的主机配置很好。

我已经使用本地 ActiveMQ 对此进行了测试，并且我的测试设置运行良好，因此消息流也没有问题。我可以看到连接的消费者和在管理控制台中创建的主题。

这是我基于Spring Documentation here的代码！

当我尝试连接客户端时，我收到连接消息，然后是同一会话的断开连接。

如果是 SSL 的问题，那么我在 Spring 文档中找不到配置 SSL 的参考。


   private static final String HOST = "b-xxxxxxxxxxxxxxxxxx.mq.aws-region.amazonaws.com";
    private static final int PORT = 61617;
    private static final String USER = "username-here";
    private static final String PASSCODE = "passcode-here";

    private ReactorNettyTcpClient<byte[]> createTcpClient() {

        return new ReactorNettyTcpClient(HOST, PORT, new StompReactorNettyCodec());
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableStompBrokerRelay()
                .setRelayHost(HOST)
                .setRelayPort(PORT)
                .setSystemLogin(USER)
                .setSystemPasscode(PASSCODE)
                .setClientLogin(USER)
                .setClientPasscode(PASSCODE)
                .setTcpClient(createTcpClient());


        registry.setApplicationDestinationPrefixes("/app")
                .setPathMatcher(new AntPathMatcher("."));

    }

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

     @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        AuthenticatedUser a = null;
        registration.interceptors(new ChannelInterceptorAdapter() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                log.info(" accessor {}", accessor);

                return message;
            }

        });

启动时记录

     Starting...
     Connecting "system" session to 
      stomp://XXXXXXXXXXXXXXXXX.mq.XXXXXX.amazonaws.com:61617
     Started.
     Tomcat started on port(s): 8080 (http) with context path ''

客户端尝试连接时记录

     accessor StompHeaderAccessor [headers={simpMessageType=CONNECT, 
     stompCommand=CONNECT, nativeHeaders={accept-version=[1.1,1.0], heart-beat=[10000,10000]}, 
     simpSessionAttributes={}, simpHeartbeat=[J@7070e97e, 
     simpSessionId=52usza4t}]

     accessor StompHeaderAccessor [headers={simpMessageType=DISCONNECT, 
     stompCommand=DISCONNECT, simpSessionAttributes={}, 
     simpSessionId=52usza4t}]

有人遇到过类似的事情吗？提前致谢 ：）

4

1 回答 1

0

我用适当的库版本解决了它，并特别与反应器发生冲突。分享对我有用的东西。

首先，我将 Spring Boot 版本从 2.0.4 升级到“2.1.6.RELEASE”

compile 'io.netty:netty-all:4.1.36.Final'

compile 'io.projectreactor:reactor-core:3.2.10.RELEASE'
compile 'io.projectreactor:reactor-net:2.0.5.RELEASE'
compile 'io.projectreactor:reactor-spring-context:2.0.7.RELEASE'
compile 'io.projectreactor.netty:reactor-netty:0.8.9.RELEASE'


// had to add this as application was failing to start complaining about jms 
context.
compile 'javax.jms:javax.jms-api:2.0.1/reactor'

compile group: 'org.springframework', name: 'spring-messaging', version: 
'5.1.8.RELEASE'
compile group: 'org.springframework', name: 'spring-jms', version: 
'5.0.6.RELEASE'


import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompReactorNettyCodec;
import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient;
import org.springframework.security.authentication.AuthenticationManager;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;

@Slf4j
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Qualifier
    private AuthenticationManager authenticationManager;

    @Autowired
    private WebSocketSecurityHandler webSocketSecurityHandler;

    @Value("${spring.config.mq.username}")
    private String userName;
    @Value("${spring.config.mq.password}")
    private String password;

    @Value("${spring.config.mq.host}")
    private String host;  // "b-xxxxxx-xxxxx-xxxx-xxxx-xxxxx-x.mq.eu-central-1.amazonaws.com";

    @Value("${spring.config.mq.port}")
    private int port;    // Amazon MQ (ActiveMQ) STOMP port = 61614


    private SocketAddress getAddress() {
        try {
            InetAddress addr = InetAddress.getByName(host);
            SocketAddress sockaddr = new InetSocketAddress(addr, port);
            return sockaddr;
        } catch (UnknownHostException e) {
            log.error("failed to connect");
        }
        return null;
    }

    private ReactorNettyTcpClient<byte[]> createTcpClient() {
        ReactorNettyTcpClient unSecured;

        unSecured = new ReactorNettyTcpClient<>(
                client -> client.addressSupplier(() -> getAddress()).secure(),
                new StompReactorNettyCodec());

        return unSecured;
    }


    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableStompBrokerRelay("/topic/", "/queue/")
                .setRelayHost(host)
                .setRelayPort(port)
                .setSystemLogin(userName)
                .setSystemPasscode(password)
                .setClientLogin(userName)
                .setClientPasscode(password)
                .setTcpClient(createTcpClient());


        registry.setApplicationDestinationPrefixes("/app");
        //.setPathMatcher(new AntPathMatcher("."));

    }

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

}
于 2019-07-03T08:42:54.873 回答