flink指定两个消费者，即先是将一个数据流保存到postgresql的一个表中，再将一个计算结果保存到另一个表中

可以通过 Flink 的 DataStream API 实现这个需求。首先，你需要定义两个 PostgreSQL 数据库的连接和表信息，可以使用 Flink 的 JDBC 连接器进行连接。然后，使用 Flink 的 DataStream API 读取数据流，将数据流分别保存到两个不同的表中。具体实现可参考以下代码示例：

// 导入必要的依赖
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.Context;
import org.apache.flink.streaming.api.scala.DataStream;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.jdbc.JdbcConnectionOptions;
import org.apache.flink.streaming.connectors.jdbc.JdbcSink;
import org.apache.flink.streaming.connectors.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.connectors.jdbc.JdbcWriter;
import org.apache.flink.streaming.connectors.jdbc.JdbcWriterFactory;
import org.apache.flink.streaming.connectors.jdbc.JdbcWriterProvider;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunction;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunctionBuilder;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunctionProvider;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunction.State;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunction.StateSerializer;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunction.TransactionContext;
import org.apache.flink.streaming.connectors.jdbc.JdbcXaSinkFunction.TransactionContextSerializer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

// 定义一个保存数据到 PostgreSQL 的 SinkFunction
class SaveToPostgreSQLSinkFunction extends RichSinkFunction<String> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private String insertStatement = "INSERT INTO table1 (id, name) VALUES (?, ?)";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 获取 PostgreSQL 数据库连接
        connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/test", "user", "password");
        // 创建 PreparedStatement
        preparedStatement = connection.prepareStatement(insertStatement);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 解析数据并设置 PreparedStatement 参数
        String[] fields = value.split(",");
        preparedStatement.setInt(1, Integer.parseInt(fields[0]));
        preparedStatement.setString(2, fields[1]);
        // 执行插入操作
        preparedStatement.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭 PreparedStatement 和数据库连接
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}

// 定义一个保存计算结果到 PostgreSQL 的 SinkFunction
class SaveResultToPostgreSQLSinkFunction extends RichSinkFunction<String> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private String insertStatement = "INSERT INTO table2 (id, result) VALUES (?, ?)";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 获取 PostgreSQL 数据库连接
        connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/test", "user", "password");
        // 创建 PreparedStatement
        preparedStatement = connection.prepareStatement(insertStatement);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 解析数据并设置 PreparedStatement 参数
        String[] fields = value.split(",");
        preparedStatement.setInt(1, Integer.parseInt(fields[0]));
        preparedStatement.setString(2, fields[2]);
        // 执行插入操作
        preparedStatement.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭 PreparedStatement 和数据库连接
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}

public class SaveToPostgreSQL {
    public static void main(String[] args) throws Exception {
        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置 Kafka 数据源
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "test");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), props);

        // 从 Kafka 中读取数据流
        DataStream<String> dataStream = env.addSource(consumer);

        // 将数据流保存到 PostgreSQL 的表1中
        dataStream.addSink(JdbcSink.sink(
            "INSERT INTO table1 (id, name) VALUES (?, ?)",
            new JdbcStatementBuilder<String>() {
                @Override
                public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                    String[] fields = s.split(",");
                    preparedStatement.setInt(1, Integer.parseInt(fields[0]));
                    preparedStatement.setString(2, fields[1]);
                }
            },
            new JdbcWriterFactory<String>() {
                @Override
                public JdbcWriter<String> createJdbcWriter(Connection connection) throws SQLException {
                    return new JdbcWriter<String>() {
                        @Override
                        public void write(Connection connection, String s) throws SQLException {
                            String[] fields = s.split(",");
                            PreparedStatement preparedStatement = connection.prepareStatement("INSERT INTO table1 (id, name) VALUES (?, ?)");
                            preparedStatement.setInt(1, Integer.parseInt(fields[0]));
                            preparedStatement.setString(2, fields[1]);
                            preparedStatement.executeUpdate();
                        }
                    };
                }
            },
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:postgresql://localhost:5432/test")
                .withDriverName("org.postgresql.Driver")
                .withUsername("user")
                .withPassword("password")
                .build()
        ));

        // 将数据流计算结果保存到 PostgreSQL 的表2中
        dataStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                String[] fields = s.split(",");
                int id = Integer.parseInt(fields[0]);
                String name = fields[1];
                int result = id * 2;
                return String.format("%d,%s,%d", id, name, result);
            }
        }).addSink(new SaveResultToPostgreSQLSinkFunction());

        // 执行任务
        env.execute("SaveToPostgreSQL");
    }
}

以上代码示例中，我们实现了两个保存数据到 PostgreSQL 的 SinkFunction，分别是 SaveToPostgreSQLSinkFunction 和 SaveResultToPostgreSQLSinkFunction。其中，SaveToPostgreSQLSinkFunction 将数据流保存到表1中，SaveResultToPostgreSQLSinkFunction 对数据流进行计算并将结果保存到表2中。在主函数中，我们使用了 Flink 的 JDBC 连接器和 JdbcSink 工具类将数据流保存到 PostgreSQL 中。同时，我们使用了 Flink 的 DataStream API 对数据流进行计算并将计算结果保存到 PostgreSQL 中。