Help With Design/Tool Choice for Distributed Application

[James Fredrickson]

Although I've taken several college classes I'm relatively new to java - as in I don't have a lot of real world experience. I'm writing a program for work (A rogue effort meant more for me than for them but it is something that is practical and gives me some experience) that requires the creation of millions of PDF files. I need some help deciding what would be the best design and the best Java toolset to use.

The current system uses a single desktop computer and takes about 2 weeks from start to finish. In addition, because there are 100's of gigabytes of data it takes about a day to transfer the files up to the network, another day to transfer the files up to our public facing web server, and then we have to copy the files to a firewire drive and ship them to the printer. This process is archaic (I didn't have anything to do with it) and causes a lot of stress for the department.

My idea:

Create a Java server that will queue the records from the database to be picked up by clients. The server will also log information received from the client such as time to process the pdf, the URL of PDF and such.
Create a java client that will run on individual workstations, grab a record from the server, create the pdf file, and upload the file to some cloud storage.
Create a managemnet console that can be used to configure certain variables and start the process.

In doing this we can use a couple of hundred idle workstations to process the pdf files, which should save us a lot of time in the creation of the PDF. In addition having the clients copy the pdf files to cloud storage would reduce the number of copies. Our internal and external website can then just reference the URL's supplied by the client.

For the PDF creation I plan to use iText and am about 60% finished with my class that creates the PDF's.
What I'm struggling with is how to best setup the interaction/coordination between the clients and server. Should I use RMI? If not what would be best in this situation. I'm not copying files to the server from the client just data (record objects from database). Also what about coordinating the traffic between the all of the clients and the server? What would be the best way to handle that? The last thing is queing the records. Shoud I read all of the records into memory on the server or should i just keep a few hundred in the queue at any one time. I'm looking for "best practice" information.

I know this is a big undertaking for a beginner but you gotta learn somehow. I would very much appreciate your input/advice. Thanks!

[James Fredrickson]

Well I'm about 90% finished with this project. I don't know why I had no responses for my question regarding the varying approaches to this. But here is what I ended up doing. I wrote a server that listens for clients to connect. Once a client connects it a new thread is created and hands the socket info over to another class that I called the client manager. The client manager loads all of the records from the database into a static ConcurrentLinkedQueue. The clientmanager has an objectoutput stream that passes the record to the client objectinput stream. I also wrote a very simple protocol that gives the state of the client. I've run several tests. Each client can do 40000 pdf's in five minutes time. One of the other nice features is that it requeues a record if the client gets disconnected. In the current system they have to start from scratch. This is working really well. With thirteen clients we should be able to generate the PDF's in 5 minutes.

So the new problem is that I'm trying to make this nice and pretty with a GUI ineterface. So i have a form that starts and stops the server. The problem is that the when I stop the server the Concurrent linked queue is not releasing it's memory, so it uses 1.2 GB of RAM. If I then click on start server button it appends the records to the queue. Here's the pertanant code:

PDF Server

package pdfserver;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Date;

public class DpdfaServer extends Thread
{
   private static int memoryTaken = 0;
   private static final int MB = 1024*1024;
   private ClientManager clientManager = null;
   private static final int NTHREADS = 100;
   private static Boolean serverStarted = false;
   private ServerSocket serverSocket = null;
   private int clientID = -1;
     
   public DpdfaServer()
   {
      
   }
   
   public static void main (String[] args)
   {
      new DpdfaServer();
   }
    @Override
   public void run()
   {
       startServer();
       return;
   }

   private void startServer()// private void start()
   {
      serverStarted = true;
      clientManager = new ClientManager();
      
      try
      {
         serverSocket = new ServerSocket(8000);
         System.out.println("Server started at " + new Date());  
      }catch(IOException ex)
      {
         System.err.println("Could not listen on port: 8000." + ex);
         System.exit(1);
      }
      
      while (serverStarted == true)
      {          
          try
          {
             Socket clientSocket = serverSocket.accept();
               
             if(clientSocket.isConnected())
             {
                ClientManager task = new ClientManager(clientSocket, 
                          clientID++);
                task.start();
                
              }
            }
            catch (IOException ex)
            {
               System.err.println("Accept failed.");
               System.exit(1);
            }
         }  
   }

   protected void stopserver()
   {
       serverStarted = false;
       serverSocket = null;
       clientManager.clearQueue();
       System.exit(0);
   }
}

Client Manager

/*
 * ClientManager Class manages and coordinates the client interaction
 */
package pdfserver;

import datatypes.DpdfProtocol;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Date;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
import datatypes.*;
import java.io.*;

public class ClientManager extends Thread
{
   private static Queue<AsrNotification> queue = 
      new ConcurrentLinkedQueue<AsrNotification>();
   private static Connection dbConnect;
   private Socket socket;
   private int clientID = -1;
   private static int numberOfRecords = 0;
   private static int numberOfClients = 0;
   private static int recordsProcessed = 0;
   private ObjectOutputStream out = null;
   //private ObjectInputStream jobinfo = null;
   private Jobinformation jobstats = null;
   private BufferedReader in = null;
   private DpdfProtocol clientStatus = new DpdfProtocol();
   private AsrNotification currentObject = null;
   private ByteArrayOutputStream[] pdfTemplates = new ByteArrayOutputStream[8];
   
   /*
    * This default constructor loads all of the mailer objects into a static
    * linked queue.
    */
   public ClientManager()
   {
      loadQueue();
   }
   /*
    * This constructor is called whenever a new client is connected to the 
    * server. It configures the socket used to communicate with the client
    */
   public ClientManager(Socket socket, int id)
   {
      setSocket(socket);
      addNumberOfClients();
      clientID = id;
      System.out.println("New Thread for client created");
      printNumberOfClients();
   }
   
   protected void clearQueue()
   {
       queue.clear();
   }
   
   //Returns the number of Clients connected to the server
   protected static void printNumberOfClients()
   {
      System.out.println("Number of clients connected: " + getNumberOfClients());
   }
   protected static int getNumberOfClients()
   {
      return numberOfClients;
   }
   //Adds one to the count of clients
   protected static void addNumberOfClients()
   {
      numberOfClients++;
   }
   //Reduces number of connected clients by one.
   protected static void reduceNumberOfClients()
   {
      numberOfClients--;
      System.out.println("Num clients reduced");
   }
   //Receives socket ID from the server and sets it for use in the client thread
   private void setSocket(Socket socket)
   {
      this.socket = socket;
   }
   /*This is the code that runs in its own thread. Manages communication 
    * between client and the server.
    */
   @Override
   public void run()
   {
      String clientStatusIn;
      
      try
      {
         out = new ObjectOutputStream
            (socket.getOutputStream());
         in = new BufferedReader(new InputStreamReader
            (socket.getInputStream()));
         
         while (!queue.isEmpty() && !socket.isClosed())
         {
            try
            {
               clientStatusIn = in.readLine();
               
               if ((clientStatusIn).equalsIgnoreCase(clientStatus.Ready()))
               {
                  currentObject = queue.poll();
                  out.writeObject(currentObject);
               }
               
               clientStatusIn = in.readLine();
                                   
               if ((clientStatusIn).equalsIgnoreCase(clientStatus.Error()))
               {
               
                  System.out.println("Requeuing: " + currentObject.getApn());
                  queue.add(currentObject);
                  reduceNumberOfClients();
                  printNumberOfClients();
                  socket.close();
                
               }
               
               clientStatusIn = in.readLine();
               
               if (clientStatusIn.equalsIgnoreCase(clientStatus.Busy()))
               {
               
               //Don't do anything yet. Will implement timer for here.
               }
                    
               
            }catch (NullPointerException ex)
            {
               System.out.println("Requeuing: " + currentObject.getApn());
                  queue.add(currentObject);
                  reduceNumberOfClients();
                  printNumberOfClients();
                  socket.close();
            }catch (SocketException ex)
            {
                System.out.println("Requeuing: " + currentObject.getApn());
                  queue.add(currentObject);
                  reduceNumberOfClients();
                  printNumberOfClients();
                  socket.close();
            }
  
         }
         in.close();
         out.close();
         socket.close();
              
      }catch(IOException ex)
      {
         System.err.println(ex);
      }

   }//end of thread run
   
   //Opens a connection with the database
   private static void openConnection()
   {      
      try
      {
         dbConnect = new ConnectDB().getDBConnection();
      }catch(SQLException ex)
      {
         Logger.getLogger(DpdfaServer.class.getName()).log(Level.SEVERE, null,
            ex);
         System.exit(0);
      }  
   }
  
   //Loads the queue with mailer objects
   protected static void loadQueue()
   {
      openConnection();
      
      System.out.println("Loading record queue...");
      try
      {
         Statement statement = dbConnect.createStatement();
         ResultSet results = statement.executeQuery("Select * from "
            + "ADMIN.VIEWITALL");
         Date start = new Date();
 
         Runtime runtime = Runtime.getRuntime();
         
         while (results.next())
         {
            AsrNotification notifyObject = new AsrNotification();
            notifyObject.setApn(results.getInt("APN"));
            notifyObject.setTra(results.getInt("TRA"));
            notifyObject.setName(results.getString("NAME"));
            notifyObject.setDoc(results.getString("DOC"));
            notifyObject.setName2(results.getString("NAME2"));
            notifyObject.setConame(results.getString("CONAME"));
            notifyObject.setStreet(results.getString("STREET"));
            notifyObject.setCitystatezip(results.getString("CITYSTATEZIP"));
            notifyObject.setForeignaddrflag(results.getString
               ("FOREIGNADDRFLAG"));
            notifyObject.setLandVal(results.getInt("LAND_VAL"));
            notifyObject.setImprVal(results.getInt("IMPR_VAL"));
            notifyObject.setSiteaddr(results.getString("SITEADDR"));
            notifyObject.setSitecitystate(results.getString("SITECITYSTATE"));
            notifyObject.setBaseyrval(results.getString("BASEYRVAL"));
            notifyObject.setFlagfield(results.getString("FLAGFIELD"));
            notifyObject.setExemptionflag(results.getString("EXEMPTIONFLAG"));
            notifyObject.setPin(results.getString("PIN"));
            notifyObject.setHoxVal(results.getInt("HOX_VAL"));
            notifyObject.setLettertype(results.getInt("LETTERTYPE"));
            notifyObject.setMailparseZipcode(results.getString
               ("MAILPARSE_ZIPCODE"));
            notifyObject.setMailparseCityname(results.getString
               ("MAILPARSE_CITYNAME"));
            notifyObject.setMailparseStatecode(results.getString
               ("MAILPARSE_STATECODE"));
            notifyObject.setMailparseZip1stfive(results.getInt
               ("MAILPARSE_ZIP_1STFIVE"));
            notifyObject.setMailparseZip2ndfour(results.getString
               ("MAILPARSE_ZIP_2NDFOUR"));
            notifyObject.setMailzipRoutenumber(results.getString
               ("MAILZIP_ROUTENUMBER"));
            notifyObject.setFbyrLand(results.getString("FBYR_LAND"));
            notifyObject.setFbyrImpr(results.getString("FBYR_IMPR"));
            notifyObject.setTotalfbyr(results.getString("TOTALFBYR"));
            notifyObject.setIsexclusion(results.getString("ISEXCLUSION"));
            notifyObject.setIsna(results.getString("ISNA"));
            notifyObject.setIspAddress(results.getString("ISP_ADDRESS"));
            notifyObject.setIspCity(results.getString("ISP_CITY"));
            notifyObject.setIspState(results.getString("ISP_STATE"));
            notifyObject.setIspZipcode(results.getString("ISP_ZIPCODE"));
            notifyObject.setIspPostnet(results.getString("ISP_POSTNET"));
            notifyObject.setIspSequence(results.getInt("ISP_SEQUENCE"));
            notifyObject.setIspOptEndorse(results.getString("ISP_OPT_ENDORSE"));
            notifyObject.setIspSacknpack(results.getString("ISP_SACKNPACK"));
            
            queue.add(notifyObject);
            notifyObject = null;
            numberOfRecords++;
            
         }
         
         dbConnect.close();
         Date finish = new Date();
         System.out.println("Start time: " + start);
         System.out.println("finish time: " + finish);
         System.out.println("Number of Records: " + numberOfRecords);
         results.close();
         statement.close();
         
      } catch (SQLException ex)
      {
         Logger.getLogger(DpdfaServer.class.getName()).log(Level.SEVERE, null,
            ex);
         System.exit(1);
      }
   }
  
   
}

This probably has to do with a thread locking the queue or perhaps the queue is still being referenced from somewhere but I cannot see it. Can someone here offer advice? Thanks!

[Harshana Dias]

I also wrote a very simple protocol that gives the state of the client

It’s a real nice effort you have done. I am studying your approach. Can you please post the protocol implementation of yours too?