Serialize/deserialize objects to and from database concurrently

Question

I have a class which runs as spring bean in the container and its purpose is data persistence. In the overriden method public void executeDataSyncOperation(DataSyncOperation[] operations) we have option to persists the data in the operations array. But the flow of data coming in is really huge and quick that we wanted to make its execution concurrent so that when fatal exception happen we don't end up with huge backlog of data which did not get persisted. One thing I should mention is that once executeDataSyncOperation method is called operations will not be modified by any other thread. So I came up with following idea;

I implemented the logic of using workManager which will get instantiated every-time the above method gets called and these workmanagers each instantiates n number of threads which then goes through the iterator created from the list which was created from the original array. Below is my implementation; can you please advice any shortcomings of this method or if there are any better methods.

CustomDataSyncEndPoint.java (class which runs as spring bean and its method is called whenever data is passed for persistence)

public class CustomDataSyncEndPoint extends SpaceSynchronizationEndpoint {

    private final Logger logger = Logger.getLogger(CustomDataSyncEndPoint.class.getName());

    private DataSource dataSource;

    private List<PersistenceWorkManager> workManagers;

    private Thread workManagerMonitor;

    private volatile boolean stopProcessing;

    private boolean isTraceLoggingEnabled;

    public CustomDataSyncEndPoint(DataSource dataSource) {
        this.dataSource = dataSource;
        workManagers = new ArrayList<PersistenceWorkManager>();
        isTraceLoggingEnabled = logger.isLoggable(Level.FINER);
    }

    @PostConstruct
    private void init() {
        stopProcessing = false;
        workManagerMonitor = new Thread(new WorkManagerMonitor(), "workManagerMonitor");
        workManagerMonitor.setDaemon(true);
        workManagerMonitor.start();
    }

    @PreDestroy
    private void destroy() {
        stopProcessing = true;
        try {
            Thread.sleep(5000);
            if (workManagerMonitor.isAlive()) {
                workManagerMonitor.interrupt();
            }
            for (PersistenceWorkManager workManager : workManagers) {
                workManager.stopProcessing();
            }
        } catch (InterruptedException e) {
            // Ignore this as we are cleaning the resources anyway
        }
    }

    @Override
    public void onTransactionSynchronization(TransactionData transactionData) {
        executeDataSyncOperation(transactionData.getTransactionParticipantDataItems());
    }

    @Override
    public void onOperationsBatchSynchronization(OperationsBatchData batchData) {
        executeDataSyncOperation(batchData.getBatchDataItems());
    }

    public void executeDataSyncOperation(DataSyncOperation[] operations) {
        if (operations == null || operations.length == 0)
            return;

        // if number of operations are greater than five then only start multiple runnables otherwise just start one runnable
        int numOfRunnables = (operations.length < 5) ? 1 : 5;

        if (isTraceLoggingEnabled)
            logger.finer("execution sync operation with " + operations.length + " dataitems");
        PersistenceWorkManager workManager = new PersistenceWorkManager(numOfRunnables, operations, dataSource);
        workManagers.add(workManager);
    }

    /**
     * WorkManagerMonitor runnable class. This class keeps polling the list of 
     * workManagers to check if they have finished processing. If so it will remove 
     * it from the list.
     * @author desai
     *
     */
    private class WorkManagerMonitor implements Runnable {

        @Override
        public void run() {
            if (isTraceLoggingEnabled)
                logger.finer("started workMangerMonitor");
            try {
                while (!stopProcessing) {
                    for (PersistenceWorkManager workManager : workManagers) {
                        if (workManager.hasFinishedPersisting()) {
                            workManagers.remove(workManager);
                        }
                    }
                    // TODO: The polling interval should be set by spring property
                    Thread.sleep(180 * 1000);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

PersistenceWorkManager.java

public class PersistenceWorkManager implements Serializable {

    private static final long serialVersionUID = 3503180330888692087L;

    private int numOfRunnables;

    private List<DataSyncOperation> datasyncOperation;

    private List<Thread> persitenceRunnableThread;

    private Iterator<DataSyncOperation> iterator;

    private DataSource dataSource;

    private volatile boolean finishedPersisting = false;

    private boolean isTraceLoggingEnabled;

    private static final Logger logger = Logger.getLogger(PersistenceWorkManager.class.getName());

    public PersistenceWorkManager(int numOfRunnables, DataSyncOperation[] datasyncOperation, DataSource dataSource) {
        this.numOfRunnables = numOfRunnables;
        this.datasyncOperation = Arrays.asList(datasyncOperation);
        iterator = this.datasyncOperation.iterator();
        this.dataSource = dataSource;
        persitenceRunnableThread = new ArrayList<Thread>();
        isTraceLoggingEnabled = logger.isLoggable(Level.FINER);
        init();
    }

    public void setFinishedPersisting(boolean finishedPersisting) {
        this.finishedPersisting = finishedPersisting;
    }

    public boolean hasFinishedPersisting() {
        return finishedPersisting;
    }

    // @PostConstruct
    public void init() {
        if (isTraceLoggingEnabled)
            logger.info("Initialising workmanager with " + numOfRunnables + " runnables");
        for (int i = 0; i < numOfRunnables; i++) {
            PersistenceRunnable persistenceRunnable = new PersistenceRunnable(this, "runnable-" + i, dataSource);
            Thread t = new Thread(persistenceRunnable, "persistenceRunnable-" + i);
            persitenceRunnableThread.add(t);
            t.start();
        }
    }

    void stopProcessing() {
        for (Thread thread : persitenceRunnableThread) {
            if (thread.isAlive() || !thread.isInterrupted()) {
                thread.interrupt();
            }
        }
    }

    public synchronized boolean hasNext() {
        return iterator.hasNext();
    }

    public synchronized DataSyncOperation next() {
        return iterator.next();
    }
}

PersistenceRunnable.java

public class PersistenceRunnable implements Runnable {

    private JdbcTemplate jdbcTemplate;

    private static final Logger logger = Logger.getLogger(PersistenceRunnable.class.getName());

    private String threadName;

    private PersistenceWorkManager workManager;

    private boolean isTraceLoggingEnabled;

    public PersistenceRunnable(PersistenceWorkManager workManager, String threadName, DataSource dataSource) {
        this.workManager = workManager;
        this.threadName = threadName;
        jdbcTemplate = new JdbcTemplate(dataSource);
        isTraceLoggingEnabled = logger.isLoggable(Level.FINER);
    }

    public JdbcTemplate getJdbcTemplate() {
        return jdbcTemplate;
    }

    @Override
    synchronized public void run() {
        if (isTraceLoggingEnabled)
            logger.finer("running runnable " + threadName);
        while (workManager.hasNext()) {
            DataSyncOperation dataSyncOperation = workManager.next();
            Object obj = dataSyncOperation.getDataAsObject();
            if (obj instanceof Data) {
                Data data = (Data) obj;
                switch (dataSyncOperation.getDataSyncOperationType()) {
                case WRITE:
                    executeWrite(data);
                    break;
                case UPDATE:
                case PARTIAL_UPDATE:
                    executeUpdateIfExists(data);
                    break;
                case REMOVE:
                    executeRemoveIfExists(data);
                    break;
                case CHANGE:
                    executeUpdateIfExists(data);
                    break;
                default:
                    logger.warning("Unsupported opeartion " + dataSyncOperation.getDataSyncOperationType().toString()
                            + " sent to mirror pu!!!");
                    break;
                }
            } else {
                logger.warning("Unsupported class sent to mirror service");
                logger.warning("class type was not Data");
                logger.warning("Class type: " + obj.getClass().getName() + ", " + obj.getClass().getSimpleName());
            }
        }
        workManager.setFinishedPersisting(true);
        workManager.stopProcessing();
        if (isTraceLoggingEnabled)
            logger.info("Finished processing from runnable " + threadName);
    }

    private void executeWrite(Data data) {
        if (isTraceLoggingEnabled)
            logger.finer(threadName + " executing write on " + data.toString());
        if (!existsInDatabase(data)) {
            String sql = "INSERT INTO data (Id, obj_name, Object) VALUES (?, ?, ?)";
            getJdbcTemplate().update(sql, new Object[] { data.getMessageUID(), data.getClass().getName(), data });
        }
    }

    private void executeRemoveIfExists(Data data) {
        if (existsInDatabase(data)) {
            if (isTraceLoggingEnabled)
                logger.finer("executing remove on " + data.toString());
            String sql = "DELETE FROM data WHERE Id = ?";
            getJdbcTemplate().update(sql, new Object[] { data.getMessageUID() });
        }
    }

    private void executeUpdateIfExists(Data data) {

        if (existsInDatabase(data)) {
            if (isTraceLoggingEnabled)
                logger.finer("executing update on " + data.toString());
            String sql = "UPDATE data SET Object = ? where Id = ?";
            getJdbcTemplate().update(sql, new Object[] { data, data.getMessageUID() });
        }
    }

    private boolean existsInDatabase(Data data) {
        String sql = "select count(*) from data where Id = ?";
        int count = getJdbcTemplate().queryForObject(sql, new Object[] { data.getMessageUID() }, Integer.class);
        if (count == 1)
            return true;
        return false;
    }
}

The reason for persisting data this way is because the class type Data is generic and all the subclasses have different and complex data types which we don't want to create separate tables for each of them.

NOTE: This needs to run on Java 7 so will not be able to use any advanced features of Java 8.

Answer 1

A more complex suggestion - you should really try to use the java.util.concurrent API. It will make your life easier. A little refactoring leads to:

public class DataPersistor implements Runnable {

    private static final Logger LOGGER = Logger.getLogger(DataPersistor.class.getName());

    private final boolean isTraceLoggingEnabled = LOGGER.isLoggable(Level.FINER);

    private final DataSyncOperation dataSyncOperation;

    private final JdbcTemplate jdbcTemplate;

    public DataPersistor(final DataSyncOperation dataSyncOperation, final DataSource dataSource) {
        this.dataSyncOperation = dataSyncOperation;
        this.jdbcTemplate = new JdbcTemplate(dataSource);
    }

    @Override
    public void run() {
        if (this.isTraceLoggingEnabled) {
            LOGGER.finer("Executing " + this.dataSyncOperation);
        }

        final Object obj = this.dataSyncOperation.getDataAsObject();
        if (obj instanceof Data) {
            final Data data = (Data) obj;
            switch (this.dataSyncOperation.getDataSyncOperationType()) {
                case WRITE:
                    executeWrite(data);
                    break;
                case UPDATE:
                case PARTIAL_UPDATE:
                    executeUpdateIfExists(data);
                    break;
                case REMOVE:
                    executeRemoveIfExists(data);
                    break;
                case CHANGE:
                    executeUpdateIfExists(data);
                    break;
                default:
                    LOGGER.warning("Unsupported operation " + this.dataSyncOperation.getDataSyncOperationType().toString() + " sent to mirror service!!!");
                    break;
            }
        } else {
            LOGGER.warning("Unsupported class sent to mirror service");
            LOGGER.warning("class type was not Data");
            LOGGER.warning("Class type: " + obj.getClass().getName() + ", " + obj.getClass().getSimpleName());
        }

        if (this.isTraceLoggingEnabled) {
            LOGGER.info("Finished executing " + this.dataSyncOperation);
        }
    }

    private void executeWrite(final Data data) {
        if (!existsInDatabase(data)) {
            String sql = "INSERT INTO data (Id, obj_name, Object) VALUES (?, ?, ?)";
            this.jdbcTemplate.update(sql, new Object[] { data.getMessageUID(), data.getClass().getName(), data });
        }
    }

    private void executeRemoveIfExists(final Data data) {
        if (existsInDatabase(data)) {
            String sql = "DELETE FROM data WHERE Id = ?";
            this.jdbcTemplate.update(sql, new Object[] { data.getMessageUID() });
        }
    }

    private void executeUpdateIfExists(final Data data) {
        if (existsInDatabase(data)) {
            String sql = "UPDATE data SET Object = ? where Id = ?";
            this.jdbcTemplate.update(sql, new Object[] { data, data.getMessageUID() });
        }
    }

    private boolean existsInDatabase(final Data data) {
        final String sql = "select count(*) from data where Id = ?";
        final int count = this.jdbcTemplate.queryForObject(sql, new Object[] { data.getMessageUID() }, Integer.class);
        return count == 1;
    }
}

and

public final class CustomDataSyncEndPoint extends SpaceSynchronizationEndpoint {

    private final Logger logger = Logger.getLogger(CustomDataSyncEndPoint.class.getName());

    private final boolean isTraceLoggingEnabled = this.logger.isLoggable(Level.FINER);

    private final Set<ExecutorService> executorServices = new HashSet<>();

    private final DataSource dataSource;

    private Thread workManagerMonitor;

    public CustomDataSyncEndPoint(final DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @PostConstruct
    private void init() {
        this.workManagerMonitor = new Thread(new ExecutorServiceMonitor(180, TimeUnit.SECONDS), "workManagerMonitor");
        this.workManagerMonitor.setDaemon(true);
        this.workManagerMonitor.start();
    }

    @PreDestroy
    private void destroy() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            Thread.interrupted();
        }

        if (this.workManagerMonitor.isAlive()) {
            this.workManagerMonitor.interrupt();
        }

        for (final ExecutorService executorService : this.executorServices) {
            executorService.shutdownNow();
        }
    }

    @Override
    public void onTransactionSynchronization(final TransactionData transactionData) {
        executeDataSyncOperation(transactionData.getTransactionParticipantDataItems());
    }

    @Override
    public void onOperationsBatchSynchronization(final OperationsBatchData batchData) {
        executeDataSyncOperation(batchData.getBatchDataItems());
    }

    public void executeDataSyncOperation(final DataSyncOperation[] operations) {
        if ((operations == null) || (operations.length == 0)) {
            return;
        }

        if (this.isTraceLoggingEnabled) {
            this.logger.finer("executing sync operation with " + operations.length + " data items");
        }

        final int numberOfThreads = (operations.length < 5) ? 1 : 5;
        final ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads, new CountingThreadFactory());
        for (final DataSyncOperation dataSyncOperation : operations) {
            executorService.submit(new DataPersistor(dataSyncOperation, this.dataSource));
        }
        executorService.shutdown(); // free up resources after all tasks are completed.
        this.executorServices.add(executorService);
    }

    /**
     * Polls the set of <code>ExecutorServices</code> and removes any that have terminated.
     * 
     * @author desai
     */
    private class ExecutorServiceMonitor implements Runnable {

        private final long sleepMillis;

        public ExecutorServiceMonitor(final long sleepTime, final TimeUnit timeUnit) {
            this.sleepMillis = timeUnit.toMillis(sleepTime);
        }

        @Override
        public void run() {
            try {
                Thread.sleep(this.sleepMillis);
            } catch (final InterruptedException e) {
                return;
            }

            for (final Iterator<ExecutorService> iterator = executorServices.iterator(); iterator.hasNext();) {
                final ExecutorService executorService = iterator.next();
                if (executorService.isTerminated()) {
                    iterator.remove();
                }
            }
        }
    }

    protected static final class CountingThreadFactory implements ThreadFactory {

        private static int threadNumber = 0;

        @Override
        public Thread newThread(final Runnable r) {
            final Thread thread = new Thread("Persistor-" + threadNumber);
            threadNumber++;
            return thread;
        }

    }
}

If you're willing to put all the DataPersistors in the same Executor, you can also get rid of the ExecutorServiceMonitor class and all its trappings. Another, heavier, option would be to use a custom extension of ThreadPoolExecutor. You can extend methods like beforeExecute() and afterExecute() for logging, and (if using multiple executors) extend terminated() to have the executor remove itself from the set.

I also agree with @h.j.k. that SLF4J would be a much better logging framework, but your hands may be tied there.

Answer 2

Some simple suggestions here...

For Java 7, you can already use type inference aka diamond operator to simplify the following such assignments:

persistenceRunnableThread = new ArrayList<>();

BTW, you have a spelling typo too: it should be persistence, not persitence in PersistenceWorkManager. Also, datasyncOperation seems to be an unnecessary instance field, as you are operating directly on the Iterator object. Therefore, you can probably just stick with:

iterator = Arrays.asList(datasyncOperation).iterator();

You have a 'guard clause' around your finer() logging lines to prevent unnecessary String concatenation, but for consistency I'll suggest that you use braces for the if statement too:

if (isTraceLoggingEnabled) {
    logger.finer("executing update on " + data.toString());
}

While you're at it, why not simply replace the java.util.Logging classes with a 'modern' logging framework like SLF4J? The methods it offers handles the String concatenation elegantly when logging is not required, so you don't even need the guard clause in the first place.

Finally, your method PersistenceRunnable.existsInDatabase(Data) can be simplified as such:

private boolean existsInDatabase(Data data) {
    String sql = "select count(*) from data where Id = ?";
    return getJdbcTemplate().queryForObject(sql, new Object[] { data.getMessageUID() }, 
                Integer.class) == 1;
}