如何仅允许单个连接(URL/端口)从Flink应用程序读写

Question

我从URL/端口读取了一些处理，然后写回URL/端口。URL/端口仅允许单个连接(您需要在需要时读写(。

flink可以读取并写入RL端口，但打开2个连接。

我已经使用了基本连接，并通过Flink

从URL/端口使用

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val data_stream = env.socketTextStream(url, port, socket_stream_deliminator, socket_connection_retries)
                         .map(x => printInput(x))
                          .writeToSocket(url, port, new SimpleStringSchema())
                         //.addSink(new SocketClientSink[String](url, port.toInt, new SimpleStringSchema))
    // execute program
    env.execute("Flink Streaming Scala API Skeleton")

理想的解决方案或唯一的解决方案是从相同的连接读取和写入，而不是创建2个分开连接

我该怎么做？

Answer 1

正如我在评论中所说的那样，您必须将连接存储在某些静态变量中，因为您的源和水槽否则不会使用相同的连接。您还必须确保使用同一classloader在同一JVM上运行源和接收器，否则您仍然具有多个连接。

我构建了此包装类别类，该类别具有原始的插座连接和该连接的读取器/作者实例。因为您的源将始终在您的水槽之前停止(这就是Flink的工作方式(，所以此类也确实重新连接了。

package example;
import java.io.BufferedReader;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.net.Socket;
public class SocketConnection implements Closeable {
    private final String host;
    private final int port;
    private final Object lock;
    private volatile Socket socket;
    private volatile BufferedReader reader;
    private volatile PrintStream writer;
    public SocketConnection(String host, int port) {
        this.host = host;
        this.port = port;
        this.lock = new Object();
        this.socket = null;
        this.reader = null;
        this.writer = null;
    }
    private void connect() throws IOException {
        this.socket = new Socket(this.host, this.port);
        this.reader = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
        this.writer = new PrintStream(this.socket.getOutputStream());
    }
    private void ensureConnected() throws IOException {
        // only acquire lock if null
        if (this.socket == null) {
            synchronized (this.lock) {
                // recheck if socket is still null
                if (this.socket == null) {
                    connect();
                }
            }
        }
    }
    public BufferedReader getReader() throws IOException {
        ensureConnected();
        return this.reader;
    }
    public PrintStream getWriter() throws IOException {
        ensureConnected();
        return this.writer;
    }
    @Override
    public void close() throws IOException {
        if (this.socket != null) {
            synchronized (this.lock) {
                if (this.socket != null) {
                    this.reader.close();
                    this.reader = null;
                    this.writer.close();
                    this.writer = null;
                    this.socket.close();
                    this.socket = null;
                }
            }
        }
    }
}

您的主类(或任何其他类(包含此类的一个实例，然后您的源和接收器都可以访问：

package example;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class Main {
    public static final SocketConnection CONNECTION = new SocketConnection("your-host", 12345);
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SocketTextStreamSource())
              .addSink(new SocketTextStreamSink());
        env.execute("Flink Streaming Scala API Skeleton");
    }
}

您的源函数看起来或多或少地像这样：

package example;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
public class SocketTextStreamSource implements SourceFunction<String> {
    private volatile boolean running;
    public SocketTextStreamSource() {
        this.running = true;
    }
    @Override
    public void run(SourceContext<String> context) throws Exception {
        try (SocketConnection conn = Main.CONNECTION) {
            String line;
            while (this.running && (line = conn.getReader().readLine()) != null) {
                context.collect(line);
            }
        }
    }
    @Override
    public void cancel() {
        this.running = false;
    }
}

和您的沉积函数：

package example;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
public class SocketTextStreamSink extends RichSinkFunction<String> {
    private transient SocketConnection connection;
    @Override
    public void open(Configuration parameters) throws Exception {
        this.connection = Main.CONNECTION;
    }
    @Override
    public void invoke(String value, Context context) throws Exception {
        this.connection.getWriter().println(value);
        this.connection.getWriter().flush();
    }
    @Override
    public void close() throws Exception {
        this.connection.close();
    }
}

请注意，我始终使用getReader()和getWriter()，因为基础插座可能在此期间已关闭。