使用Akka Http作为WebSocket服务器和客户端，添加对数据的提取和处理逻辑。

示例代码：

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.http.javadsl.ConnectHttp;
import akka.http.javadsl.Http;
import akka.http.javadsl.model.*;
import akka.http.javadsl.model.headers.RawHeader;
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.*;

import java.util.concurrent.CompletionStage;

public class WebSocketClient {

    public static void main(String[] args) {
        final ActorSystem system = ActorSystem.create();
        final ActorMaterializer materializer = ActorMaterializer.create(system);

        final String authorizationToken = "";

        final Flow flow =
                Flow.fromSinkAndSource(
                        Sink.foreach(System.out::println),
                        Source.maybe());

        final Uri wsUri = Uri.create("ws://localhost:8080/ws");

        final CompletionStage response = Http.get(system)
                .singleWebSocketRequest(
                        WebSocketRequest.create(wsUri.withQuery(Query.create("token=" + authorizationToken))),
                        flow,
                        ConnectHttp.toHost("localhost", 8080),
                        mat);

        response.whenComplete((wsUpgradeResponse, throwable) -> {
            if (throwable != null) {
                System.out.println("WebSocket connection failure: " + throwable.getMessage());
            } else {
                System.out.println("Connected to WebSocket server with HTTP response status " +
                        wsUpgradeResponse.response().status());
            }
        });
    }
}