Part 1 - Implementing Multi Threaded Batching Framework In Java

In our project we have currently a requirement to migrate(upgrade) data In Some Olden Version Of Release To Latest Version.

In olden version our schema and design of database tables are quite different when compared to current release.

So as various releases have progressed our java code has also been changed quite drastically.

Now recently we had one requirement to write a Java API which when given a application number will convert the data for the application number from older version to latest version schema tables.

As there are 1000's of applications which need to be converted hence we are told to write a multi-threaded batching framework which will run the conversion at once.

Now let us look into how we implemented in our project.

Firstly thanks to Our Developer Roy Harrington who has implemented this framework.

Batching Framework contains 4 primary classes

a) Batch.java
b) BatchProcess abstract class
c) BatchProcessor abstract class
d) BatchSet abstract class

Code and the description for each method is given as below.

BatchSet.java


public abstract class BatchSet
{
  private Batch batch;

  private boolean finished;

  private long count;

  public BatchSet(Batch batch)
  {
    this.batch = batch;
    finished = false;
  }

  public abstract void open();

  public abstract void close();

  public abstract long total();

  protected abstract Object nextRow();

  /**
   * Returns the next object from the BatchSet
   * 
   * @return
   */
  public synchronized Object next()
  {
    try
    {
      if (finished) return null;
      Object row = nextRow();
      if (null == row)
        finished = true;
      else
        count++;
      return row;
    }
    catch (RuntimeException re)
    {
      // things are screwed if this has happened, so mark as finished
      batch.setBatchSuccessful(false);
      finished = true;

      return null; // indicate nothing more
    }
  }

  /**
   * Returns the total number of rows processed so far
   * 
   * @return
   */
  public synchronized long count()
  {
    return count;
  }

}


BatchProcessor.java

public abstract class BatchProcessor implements Runnable
{
  private Batch batch;

  private BatchSet set;

  public BatchProcessor(Batch batch, BatchSet set)
  {
    this.batch = batch;
    this.set = set;
  }

  public void run()
  {
    Object next = null;

    try
    {
      try
      {
        begin();
      }
      catch (Exception e)
      {
        // mark batch as having an error
        batch.setBatchSuccessful(false);

        recordError(e);

        return;
      }

      while (null != (next = set.next()))
      {
        try
        {
          execute(next);
          continue;
        }
        catch (Throwable t)
        {
          // mark batch as having an error
          batch.setBatchSuccessful(false);

          recordError(t);
        }
      }

      try
      {
        end();
      }
      catch (Exception e)
      {
        // mark batch as having an error
        batch.setBatchSuccessful(false);
        return;
      }
    }
    finally
    {
      batch.finished(this);
    }

  }

  public abstract void begin() throws Exception;

  public abstract void execute(Object obj) throws Exception;

  public abstract void end() throws Exception;

}


BatchProcess.java


public abstract class BatchProcess
{
  private static final String METRIC_DELAY_KEY = "metricDelay";

  private static final String METRIC_DELAY_DEFAULT = "5000";

  private static final String THREADS_KEY = "threads";;

  private static final String THREADS_DEFAULT = "5";

  /**
   * Must implement this to do the work
   *
   */
  protected abstract void goProcess();

  /**
   * Get the metric delay for the class
   *
   */
  protected final int getMetricDelay()
  {
    try
    {
      ResourceHandler resourceHandler = ResourceHandler.getInstance();

      return Integer.valueOf(resourceHandler.getValue(getClass(), METRIC_DELAY_KEY, METRIC_DELAY_DEFAULT).trim()).intValue();
    }
    catch (NumberFormatException e)
    {
      return 5000;
    }
  }

  /**
   * Get the maximum number of threads for this class
   *
   */
  protected final int getMaxThreads()
  {
    try
    {
      ResourceHandler resourceHandler = ResourceHandler.getInstance();
      return Integer.valueOf(resourceHandler.getValue(getClass(), THREADS_KEY, THREADS_DEFAULT).trim()).intValue();
    }
    catch (NumberFormatException e)
    {
      return 5;
    }
  }

  /**
   * Check for the process running, write an entry to say we are running, call
   * goProcess() on subclass, clear down process table
   */
  public final void go()
  {
    String processName = getClass().getName();

    try
    {
      // Execute actual Process
      goProcess();
    }
    catch (RuntimeException re)
    {
      System.exit(1);
    }
  }
}

.. TO Be Continued In next article