从套接字读取flink中的字符串数据流,而不使用netcat服务器

qmelpv7a  于 2021-06-25  发布在  Flink
关注(0)|答案(1)|浏览(397)

我有一个案例场景,其中我有一个流生成器客户端,它生成多个流,合并它们并将其发送到socket,我希望flink程序将其作为服务器进行侦听。正如我们所知,必须首先打开服务器,以便它能够侦听客户机请求。我试着用下面给出的代码来做同样的事情

public static void main(String[] args)  throws Exception {

    //setting the envrionment variable as StreamExecutionEnvironment
      StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.setParallelism(1);

        DataStream<String> stream1 = environment.socketTextStream("localhost", 9000);
        stream1.print();

                //start the execution
     environment.execute(" Started the execution ");

}// main

流生成器作为客户端的代码如下所示

DataStream<Event> stream1  = envrionment
                .addSource(new EventGenerator(2,60,1,1,100, 200 ))
                .name("stream 1")
                .setParallelism(parallelism_for_stream_rr);

    DataStream<Event> stream2  = envrionment
            .addSource(new EventGenerator(3,60,1,2,10, 20 ))
            .name("stream 2")
            .setParallelism(parallelism_for_stream_rr);

    DataStream<Event> stream3  = envrionment
            .addSource(new EventGenerator(5,60,1,3,30, 40 ))
            .name("stream 3")
            .setParallelism(parallelism_for_stream_rr);

    DataStream<Event> merged = stream1.union(stream2,stream3);

    merged.print();

            // sending data to Mobile Cep via socket

            merged.map(new MapFunction<Event, String>() {

                @Override
                public String map(Event event) throws Exception {
                    String tuple = event.toString();

                    return tuple + "\n";

                }
            }).writeToSocket("localhost", 9000, new SimpleStringSchema() );

问题#1:问题是客户端代码只有在启动netcat服务器时才起作用,但是netcat服务器不会转发数据流。如果netcat服务器没有启动,客户端代码会说它无法建立连接
问题2:如果netcat服务器没有启动,flink程序就不会执行

Caused by: java.net.ConnectException: Connection refused (Connection refused)
at java.net.PlainSocketImpl.socketConnect(Native Method)
at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)

我知道一个可能的解决方案是在flink程序中生成流,但是我希望通过socket接收流。
提前谢谢~

nfg76nw0

nfg76nw01#

flink的套接字源和接收器都不会启动tcp服务器并等待传入连接。它们都是针对已启动的tcp服务器进行连接的客户端。这也是你必须开始的原因 netcat 在启动作业之前。如果您想对套接字进行读写,那么就必须编写一个tcp服务器,该服务器可以缓冲传入的数据,并在客户端连接到它时转发它们。

相关问题