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.