Spring Boot GraphQL客户端适配graphql-ws协议连接问题

问题根源 #

你遇到的问题核心是GraphQL WebSocket存在两种不兼容的协议标准：

• Spring Boot Starter GraphQL的WebSocketGraphQlClient默认使用新版graphql-transport-ws协议
• 你的服务器使用的是旧版graphql-ws（又称subscriptions-transport-ws）协议

两者的握手子协议标识、消息类型格式完全不同，导致握手失败、消息解码异常。

解决方案 #

需要手动配置客户端适配graphql-ws协议，包括指定正确的子协议、自定义消息编解码逻辑处理旧协议的消息类型。

1. 配置WebSocket客户端子协议 #

构建WebSocketGraphQlClient时，明确指定握手子协议为graphql-ws：

import org.springframework.graphql.client.WebSocketGraphQlClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import java.util.Collections;

StandardWebSocketClient webSocketClient = new StandardWebSocketClient();

WebSocketGraphQlClient client = WebSocketGraphQlClient.builder(
        "ws://172.16.60.198:8081/subscriptions", webSocketClient)
        .webSocketConfigurer(session -> 
            session.getHandshakeHeaders().put(
                "Sec-WebSocket-Protocol", 
                Collections.singletonList("graphql-ws")
            )
        )
        .build();

2. 自定义消息编解码适配graphql-ws协议 #

默认的GraphQlWebSocketMessage仅支持graphql-transport-ws的消息类型（如ping/pong），无法解析graphql-ws的ka（心跳）等类型，需要替换默认消息转换器：

步骤1：实现graphql-ws协议的消息模型

创建对应旧协议的消息类，覆盖所有消息类型：

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
    @JsonSubTypes.Type(value = ConnectionInitMessage.class, name = "connection_init"),
    @JsonSubTypes.Type(value = ConnectionAckMessage.class, name = "connection_ack"),
    @JsonSubTypes.Type(value = SubscribeMessage.class, name = "subscribe"),
    @JsonSubTypes.Type(value = NextMessage.class, name = "next"),
    @JsonSubTypes.Type(value = ErrorMessage.class, name = "error"),
    @JsonSubTypes.Type(value = CompleteMessage.class, name = "complete"),
    @JsonSubTypes.Type(value = KeepAliveMessage.class, name = "ka")
})
@JsonIgnoreProperties(ignoreUnknown = true)
public abstract class GraphqlWsMessage {
    private String type;

    public String getType() { return type; }
    public void setType(String type) { this.type = type; }
}

class ConnectionInitMessage extends GraphqlWsMessage {
    private Object payload;
    public Object getPayload() { return payload; }
    public void setPayload(Object payload) { this.payload = payload; }
}

class ConnectionAckMessage extends GraphqlWsMessage {}

class SubscribeMessage extends GraphqlWsMessage {
    private String id;
    private Object payload;
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    public Object getPayload() { return payload; }
    public void setPayload(Object payload) { this.payload = payload; }
}

class NextMessage extends GraphqlWsMessage {
    private String id;
    private Object payload;
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    public Object getPayload() { return payload; }
    public void setPayload(Object payload) { this.payload = payload; }
}

class ErrorMessage extends GraphqlWsMessage {
    private String id;
    private Object payload;
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    public Object getPayload() { return payload; }
    public void setPayload(Object payload) { this.payload = payload; }
}

class CompleteMessage extends GraphqlWsMessage {
    private String id;
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
}

class KeepAliveMessage extends GraphqlWsMessage {}

步骤2：自定义消息编解码器

创建适配旧协议的编解码器，替换默认实现：

import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.graphql.client.WebSocketGraphQlClient;
import org.springframework.graphql.client.WebSocketGraphQlTransport;
import org.springframework.graphql.support.DefaultGraphQlWebSocketMessageCodec;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.web.socket.WebSocketSession;

public class GraphqlWsTransportCustomizer implements WebSocketGraphQlClient.TransportCustomizer {

    @Override
    public void customize(WebSocketGraphQlTransport transport, WebSocketSession session) {
        Encoder<GraphqlWsMessage> encoder = new Jackson2JsonEncoder();
        Decoder<GraphqlWsMessage> decoder = new Jackson2JsonDecoder();
        DefaultGraphQlWebSocketMessageCodec codec = new DefaultGraphQlWebSocketMessageCodec(encoder, decoder);
        
        transport.setMessageCodec(codec);
    }
}

步骤3：应用自定义Transport配置

在构建客户端时添加自定义配置：

WebSocketGraphQlClient client = WebSocketGraphQlClient.builder(
        "ws://172.16.60.198:8081/subscriptions", webSocketClient)
        .webSocketConfigurer(session -> 
            session.getHandshakeHeaders().put(
                "Sec-WebSocket-Protocol", 
                Collections.singletonList("graphql-ws")
            )
        )
        .transportCustomizer(new GraphqlWsTransportCustomizer())
        .build();

3. 处理协议初始化流程 #

graphql-ws要求客户端先发送connection_init消息，服务器返回connection_ack后才能发送订阅请求：

// 发送连接初始化消息
client.execute("{\"type\":\"connection_init\"}")
        .subscribe(response -> {
            System.out.println("连接已确认: " + response);
        }, error -> {
            System.err.println("连接初始化失败: " + error);
        });

// 发送订阅请求
client.document("subscription { yourSubscriptionField }")
        .subscribe(response -> {
            System.out.println("收到订阅数据: " + response.getData());
        }, error -> {
            System.err.println("订阅出错: " + error);
        });

注意事项 #

• 确保Spring Boot Starter GraphQL版本至少为1.1.x，旧版本API可能不兼容
• 若服务器需要认证，可在connection_init的payload中携带token等凭证
• 服务器发送的ka心跳消息无需客户端回应，自定义解码器会自动识别处理

内容的提问来源于stack exchange，提问作者altindalorcun