侦探不适用于 websocket 客户端

Question

我有一个微服务连接到 WebSocket 服务器，并从服务器接收消息。我想在这个应用程序中集成 Sleuth，因为这是我的一组微服务的网关。但是当我连接到WebSocket服务器并开始接收消息时，我无法在日志中traceId，spanid。我期待的是这样的：

2018-08-09 11:26:20.222  INFO [kumarman-betsync-adapter-service-v2,0a011711c2470359,0a011711c2470359,true] 7230 --- [ient-SecureIO-2] c.b.b.a.service.BetSyncEventIdExtractor  : SubscribeResponse feed received for eventId=7519016
2018-08-09 11:26:20.224  INFO [kumarman-betsync-adapter-service-v2,0a011711c2470359,2b028dadad5c440d,true] 7230 --- [ient-SecureIO-2] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncEventTopic'

但我得到了：

2018-08-09 11:26:20.222  INFO [kumarman-betsync-adapter-service-v2,,,] 7230 --- [ient-SecureIO-2] c.b.b.a.service.BetSyncEventIdExtractor  : SubscribeResponse feed received for eventId=7519016
2018-08-09 11:26:20.224  INFO [kumarman-betsync-adapter-service-v2,,,] 7230 --- [ient-SecureIO-2] c.b.b.a.k.message.KafkaMessageSender     : sent payload to topic='betsyncEventTopic'

我的代码如下：

  @Service
    @Slf4j
    @AllArgsConstructor
    public class BetsyncWebsocketAdapter {

        private final BetsyncWebSocketHandler betsyncWebSocketHandler;

        public void startNewConnection() {
                try {
                    webSocketSession = new StandardWebSocketClient().doHandshake(
                            betsyncWebSocketHandler,
                            new WebSocketHttpHeaders(),
                            new URI(getBetsyncConnectionUrl()))
                            .get(KafkaAdminClient.MAX_TIME_LIMIT_TO_CONNECT, TimeUnit.SECONDS);
                } catch (Exception e) {
                    log.error("exception occurred", e);
                }
        }

        /**
         * This will start the new connection after server start
         */
        @EventListener(ApplicationReadyEvent.class)
        public void onServerStart() {
            startNewConnection();
        }
    }

@Slf4j
@Service
@AllArgsConstructor
public class BetsyncWebSocketHandler extends TextWebSocketHandler {

    private static StringBuilder betSyncMessage = new StringBuilder();

    private final LoginRequestBuilder loginRequestBuilder;
    private final BetsyncMessageProcessing betsyncMessageProcessing;

    @Override
    public void afterConnectionEstablished(final WebSocketSession session) throws IOException {
        session.sendMessage(new TextMessage(loginRequestBuilder.getLoginRequestDto()));
    }

    @Override
    protected void handleTextMessage(final WebSocketSession session, final TextMessage message) {
        betSyncMessage.append(message.getPayload());
        if (message.isLast()) {
            betsyncMessageProcessing.process(betSyncMessage.toString());
            betSyncMessage = new StringBuilder();
        } else {
            log.debug("we have received a partial message starting with {}", getFirstXCharacter(message.getPayload(), 50));
        }
    }

    @Override
    protected void handlePongMessage(final WebSocketSession session, final PongMessage message) {
        log.info("Response for Ping message received from the server");
    }

    @Override
    public boolean supportsPartialMessages() {
        return true;
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.error("server connection closed");
    }
}

我们的 pom 条目是：

<dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-core</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-sleuth</artifactId>
        </dependency>

<dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring.boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
        <spring.boot.version>2.0.3.RELEASE</spring.boot.version>
        <spring-cloud.version>Finchley.RELEASE</spring-cloud.version>

Answer 1

您正在自己创建套接字处理

 webSocketSession = new StandardWebSocketClient().doHandshake(
                            betsyncWebSocketHandler,
                            new WebSocketHttpHeaders(),
                            new URI(getBetsyncConnectionUrl()))
                            .get(KafkaAdminClient.MAX_TIME_LIMIT_TO_CONNECT, TimeUnit.SECONDS);

当您直接使用 Spring Integration 组件时，Spring Cloud Sleuth 支持 WebSockets。由于您是手动执行操作，因此您必须确保跟踪上下文得到传播并且您可以自己阅读。