如何实现异步处理

Question

我希望在我的Java程序中传递异步消息，因此在第一步中，它应该持续监视DB中某些表的更改。当有新的传入消息时，它应该显示它。只要应用程序在运行，这应该是一个重复的过程。

我可以知道如何继续这下面的代码，其中有轮询方法必须保持无限调用自己每6秒，也应该在数据库中找到新的传入消息。

下面是代码片段:

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;
}
}

这里序列id是表中的唯一id，它随着新传入消息的到来而不断增加。

p。学生:"善意的请求:请给出负面分数(大拇指不及格)的理由。"这样我就可以清楚地解释我的问题

Answer 1

放到while(true)循环中。

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);
   }
}

希望您已经将Runnable作为一个新线程启动。

对于新的和更新的消息，您需要一个字段，例如数据库中的"last_update"。然后，当您检查新消息时，您需要将获取新消息的SQL语句更改为如下内容:" where last_update > $lastCheckedDate "，其中设置了lastCheckedDate。

也许你也想读一些关于Java并发的东西:http://docs.oracle.com/javase/tutorial/essential/concurrency/

Answer 2

放入while循环是一种方法，但我认为最好避免这样做(有很多事情会弄乱，如事务等)。

如果您确实需要做这种重复的事情，请考虑使用调度器。春天3。xdo有内置的调度器，或者你也可以使用Quartz。

更好的方法是避免这样的轮询。是否有可能在数据更新时将消息放入JMS队列，以便在JMS队列中出现这样的消息时调用您的逻辑(通过消息驱动bean) ?(这只是一种可能的方法，还有很多类似的方法)