Flink CDC只跑一个就同步挺快的，是不是哪里没配置好？自定义sink

Flink CDC只跑一个就同步挺快的，是不是哪里没配置好？自定义sink private Connection connection;

@Override
public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    Class.forName("oracle.jdbc.driver.OracleDriver");
    connection = DriverManager.getConnection("jdbc:oracle:thin:@127.0.0.1:1521/orcl", "scott", "123456");
}

@Override
public void invoke(DataSyncBean value, Context context) {
    System.out.println("CustomSink:" + value);
    PreparedStatement statement = null;
    StringBuilder sql = new StringBuilder();
    //数据处理
    try {
        switch (value.getOperation()) {
            case "1":
                sql.append("insert into ");
                sql.append(value.getTable()).append("(").append(StringUtils.join(value.getFields(), ",")).append(") VALUES(");
                for (int i = 0; i < value.getValues().size(); i = i + 1) {
                    sql.append("?");
                    if (i < value.getValues().size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(")");

                statement = connection.prepareStatement(sql.toString());
                for (int i = 0; i < value.getValues().size(); i++) {
                    statement.setObject(i + 1, value.getValues().get(i));
                }
                break;
            case "2":
                sql.append("update ");
                sql.append(value.getTable()).append(" set ");
                for (int i = 0; i < value.getValues().size(); i = i + 1) {
                    if (!value.getFields().get(i).equals(value.getPk())) {
                        sql.append(value.getFields().get(i));
                        sql.append("=?");
                    }
                    if (!value.getFields().get(i).equals(value.getPk()) && i < value.getValues().size() - 1) {
                        sql.append(",");
                    }
                }
                sql.append(" where ");
                sql.append(value.getPk()).append("=?");

                statement = connection.prepareStatement(sql.toString());
                int index = 1;
                for (int i = 0; i < value.getValues().size(); i = i + 1) {
                    if (value.getFields().get(i).equals(value.getPk())) {
                        statement.setObject(value.getValues().size(), value.getPkValue());
                    } else {
                        statement.setObject(index, value.getValues().get(i));
                        index = index + 1;
                    }
                }

                break;

            case "3":
                sql.append("delete from ");
                sql.append(value.getTable()).append(" where ");
                sql.append(value.getPk()).append("=?");

                statement = connection.prepareStatement(sql.toString());
                statement.setObject(1, value.getPkValue());

                break;
        }

        if (statement != null) {
            statement.execute();
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        System.out.println(sql);
    }
}

@Override
public void close() throws Exception {
    super.close();
    if (connection != null) {
        connection.close();
    }
}