How to achieve asynchronous processing

Question

I would like to have an asynchronous message passing in my Java program ,so for the first step it should continuously monitor the changes of some table in the DB. And when there are new incoming messages, it should display it. This is should be repetitive process, as long as the application is running.

May i Know how to proceed this for the following code, which has the polling method in it which has to keep on calling itself infinitely every 6 seconds and also should find the new incoming messages in the database.

here is the code snippet:

public class PollingSynchronizer implements Runnable {

private Collection<KPIMessage> incomingMessages;
private Connection dbConnection;


/**
 * Constructor. Requires to provide a reference to the KA message queue
 * 
 * @param incomingMessages reference to message queue
 * 
 */
   public PollingSynchronizer(Collection<KpiMessage> incomingMessages, Connection dbConnection) {
    super();
    this.incomingMessages = incomingMessages;
    this.dbConnection = dbConnection;
}

private int sequenceId;

public int getSequenceId() {
    return sequenceId;
}

public void setSequenceId(int sequenceId) {
    this.sequenceId = sequenceId;
}



@Override
/**
 * The method which runs Polling action and record the time at which it is done
 * 
 */
public void run() {
    try {


           incomingMessages.addAll(fullPoll());
            System.out.println("waiting 6 seconds");

            //perform this operation in a loop
            Thread.sleep(6000);

    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } 
    Date currentDate = new Date();
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSSSS");
//  System.out.println(sdf.format(currentDate) + " " + msg);
}

/**
 * Method which defines polling of the database and also count the number of Queries
 * @return 
 * @throws Exception
 */
public List<KpiMessage> fullPoll() throws Exception {

//  int sequenceID = 0;
    Statement st = dbConnection.createStatement();

    ResultSet rs = st.executeQuery("select * from msg_new_to_bde where ACTION = 804 order by SEQ DESC");
        List<KpiMessage> pojoCol = new ArrayList<KpiMessage>();
        while (rs.next()) {
            KpiMessage filedClass = convertRecordsetToPojo(rs);
            pojoCol.add(filedClass);
        }

        return pojoCol;
        }

/**
 * Converts a provided record-set to a {@link KpiMessage}.
 * 
 * The following attributes are copied from record-set to pojo:
 * 
 * <ul>
 * <li>SEQ</li>
 * <li>TABLENAME</li>
 * <li>ENTRYTIME</li>
 * <li>STATUS</li>
 * </ul>
 * 
 * @param rs
 *            the recordset to convert
 * @return the converted pojo class object
 * @throws SQLException
 *             if an sql error occurrs during processing of recordset
 */
private KpiMessage convertRecordsetToPojo(ResultSet rs) throws SQLException {

    KpiMessage msg = new KpiMessage();
    int sequence = rs.getInt("SEQ");
    msg.setSequence(sequence);
    int action = rs.getInt("ACTION");
    msg.setAction(action);
    String tablename = rs.getString("TABLENAME");
    msg.setTableName(tablename);
    Timestamp entrytime = rs.getTimestamp("ENTRYTIME");
    Date entryTime = new Date(entrytime.getTime());
    msg.setEntryTime(entryTime);
    Timestamp processingtime = rs.getTimestamp("PROCESSINGTIME");
    if (processingtime != null) {
        Date processingTime = new Date(processingtime.getTime());
        msg.setProcessingTime(processingTime);
    }
    String keyInfo1 = rs.getString("KEYINFO1");
    msg.setKeyInfo1(keyInfo1);
    String keyInfo2 = rs.getString("KEYINFO2");
    msg.setKeyInfo2(keyInfo2);
    return msg;
}
}

Here the sequence id is the unique id in the table which keeps on increasing as new incoming messages arrive.

P.S : "Kind request : Pls give a reason for giving negative marks (thumbs down) . So that I can explain my question clearly"

Answer 1

Simple put it in a while(true) loop.

public void run() {
    while(true){
        try {


               incomingMessages.addAll(fullPoll());
                System.out.println("waiting 6 seconds");

                //perform this operation in a loop
                Thread.sleep(6000);

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } 
        Date currentDate = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSSSS");
    //  System.out.println(sdf.format(currentDate) + " " + msg);
   }
}

Hopefully you already started the Runnable as a new thread.

For new and updated messages you need a field, for instance "last_update" within the database. Than you need to alter your SQL statement for getting the new message to something like: "where last_update > $lastCheckedDate", where lastCheckedDate is set, whenever you check for new messages.

MAybe you also want to read something about concurrency in Java: http://docs.oracle.com/javase/tutorial/essential/concurrency/

Answer 2

putting in a while loop is one way, but I think it is better to avoid doing in such approach (there are lots of things to mess up, like transaction etc).

If you really need to do such kind of repetitive thing, consider using a scheduler. Spring 3.x do has scheduler built-in, or you can use Quartz too.

Even better way is to avoid polling like this. Is it possible to put a message in JMS queue when the data is updated, so that your logic will be invoked (through a message driven bean) once there is such message in JMS queue? (just one possible way, there are lots of similar way to do)