FWD » Core Development » Database

import com.mchange.v2.c3p0.ComboPooledDataSource;

import java.sql.*;

import java.util.*;

import java.util.concurrent.*;

public class Test {

    private static final ComboPooledDataSource dataSource = new ComboPooledDataSource();

    static {

        try {

            dataSource.setDriverClass("org.postgresql.Driver");

            dataSource.setJdbcUrl("jdbc:postgresql://localhost:5433/fwd");

            dataSource.setUser("fwd_admin");

            dataSource.setPassword("admin");

            dataSource.setMinPoolSize(10);

            dataSource.setMaxPoolSize(100);

        } catch (Exception e) {

            throw new RuntimeException("Error setting up datasource", e);

        }

    }

    private static void async(String query) throws SQLException, InterruptedException {

        ExecutorService executor = Executors.newSingleThreadExecutor();

        try (Connection conn = dataSource.getConnection()) {

            conn.setAutoCommit(false);

            try (Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {

                stmt.setFetchSize(50);

                try (ResultSet rs = stmt.executeQuery(query)) {

                    ResultSetMetaData meta = rs.getMetaData();

                    int columnCount = meta.getColumnCount();

                    while (rs.next()) {

                        Map<String, Object> row = extractRow(rs, meta, columnCount);

                        // Submit business logic for execution in single thread (order preserved)

                        executor.submit(() -> {

                            try {

                                executeBusinessLogic(row);

                            } catch (InterruptedException e) {

                                throw new RuntimeException(e);

                            }

                        });

                    }

                }

            }

        }

        // Shut down and wait for all tasks to complete

        executor.shutdown();

        executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

    }

    private static void sync(String query) throws SQLException, InterruptedException, ExecutionException {

        try (Connection conn = dataSource.getConnection();

             Statement stmt = conn.createStatement()) {

            try (ResultSet rs = stmt.executeQuery(query)) {

                ResultSetMetaData meta = rs.getMetaData();

                int columnCount = meta.getColumnCount();

                while (rs.next()) {

                    Map<String, Object> row = extractRow(rs, meta, columnCount);

                    executeBusinessLogic(row);

                }

            } catch (InterruptedException e) {

                throw new RuntimeException(e);

            }

        }

    }

    private static void executeBusinessLogic(Map<String, Object> row) throws InterruptedException {

        Thread.sleep(1);

    }

    private static Map<String, Object> extractRow(ResultSet rs, ResultSetMetaData meta, int columnCount) throws SQLException {

        Map<String, Object> row = new HashMap<>();

        for (int i = 1; i <= columnCount; i++) {

            row.put(meta.getColumnLabel(i), rs.getObject(i));

        }

        return row;

    }

    public static void main(String[] args) throws SQLException, ExecutionException, InterruptedException {

        double sync = System.currentTimeMillis();

        sync("SELECT *, pg_sleep(0.000003) FROM employees;");

        sync = System.currentTimeMillis() - sync;

        System.out.println(sync);

    }

}