我有一个cassandra数据库，它必须在我的flink程序中从socket-like-steam接收数据进行流处理。因此，我编写了一个简单的客户端程序，从cassandra读取数据并将数据发送到socket；另外，我在serverbase中编写了flink程序。事实上，我的客户机程序很简单，不使用任何flink指令；它只发送一个字符串格式的cassandra行到socket，服务器必须接收该行。首先，我运行flink程序来监听客户机，然后运行客户机程序。客户端从服务器接收到此流（因为服务器发送数据流数据，而客户端无法正确接收）：
hi客户端org.apache.flink.streaming.api.datastream。datastreamsource@68c72235
之后，两个程序都保持运行，没有发送和接收任何数据，没有错误。
flink程序如下：公共类wordcount\u in\u cassandra{

private static int myport=9999;
 private static String hostname="localhost";
 //static ServerSocket variable
 private static ServerSocket server;
 private static int count_row=0;

 public static void main(String[] args) throws Exception {
 // Checking input parameters
 final ParameterTool params = ParameterTool.fromArgs(args);
 // set up the execution environment
 final StreamExecutionEnvironment env = 
 StreamExecutionEnvironment.getExecutionEnvironment();

 //create the socket server object
    server = new ServerSocket(myport);
 // make parameters available in the web interface
    env.getConfig().setGlobalJobParameters(params);

    while (true){
        System.out.println("Waiting for client request");
        //creating socket and waiting for client connection
        Socket socket = server.accept();
        DataStream<String> stream = env.socketTextStream(hostname, 
        myport);

        stream.print();

        //write object to Socket
        oos.writeObject("Hi Client " + stream.toString());
        oos.close();
        socket.close();

        // parse the data, group it, window it, and aggregate the 
        counts
    DataStream<Tuple2<String, Long>> counts = stream
                .flatMap(new FlatMapFunction<String, Tuple2<String, 
    Long>>() {
                    @Override
            public void flatMap(String value, 
     Collector<Tuple2<String, Long>> out) {
                        // normalize and split the line
           String[] words = value.toLowerCase().split("\\W+");

                        // emit the pairs
             for (String word : words) {

                if (!word.isEmpty()) {
                   out.collect(new Tuple2<String, Long>(word, 1L));
                            }
                        }
                    }
                })
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);

        // emit result
        if (params.has("output")) {
            counts.writeAsText(params.get("output"));
        } else {
            System.out.println("Printing result to stdout. Use -- 
            output to specify output path.");

            counts.print();
        }

        //terminate the server if client sends exit request
        if (stream.equals("exit")){
            System.out.println("row_count : "+count_row);
            break;
        }

        // execute program
        env.execute("Streaming WordCount");
    }//while true
    System.out.println("Shutting down Socket server!!");
    server.close();
     }//main
   }

客户端程序如下所示：

public class client_code {
private static Cluster cluster = 
  Cluster.builder().addContactPoint("127.0.0.1")
 .withPort(9042).build();
private static Session session = cluster.connect("mar1");

 public static void main(String[] args) throws UnknownHostException, 
   IOException, ClassNotFoundException, InterruptedException {
    String serverIP = "localhost";
    int port=9999;
    Socket socket = null;
    ObjectOutputStream oos = null;
    ObjectInputStream ois = null;

    ResultSet result = session.execute("select * from tlbtest15");
    for (Row row : result) {
        //establish socket connection to server
        socket = new Socket(serverIP, port);
        //write to socket using ObjectOutputStream
        oos = new ObjectOutputStream(socket.getOutputStream());
        System.out.println("Sending request to Socket Server");

        if (row==result) oos.writeObject("exit");
        else oos.writeObject(""+row+"");
        //read the server response message
        ois = new ObjectInputStream(socket.getInputStream());
        String message = (String) ois.readObject();
        System.out.println("Message: " + message);
        //close resources
        ois.close();
        oos.close();
        Thread.sleep(100);
    }

    cluster.close();
 }
}

你能告诉我怎样才能解决我的问题吗？
任何帮助都将不胜感激。