0

如何处理一个正在轮询而另一个应在处理后更新新传入数据的线程池。程序执行存在于具有 main 方法和线程池的控制器类中：

主类控制器

   public static void main(String[] args) throws InterruptedException {
    RunnableController controller = new RunnableController();
    Accumulator acque = new Accumulator();
        controller.initializeDb();
        controller.initialiseThreads(acque);
        controller.initialUpdate(acque);    

}

Polling 类的 Run 方法：

     public void run() {
    int seqId = 0;
    List<KpiMessage> list = null;
    while(true) {
        try{
            list = fullPoll(seqId);
            if (!list.isEmpty()) {
            accumulator.manageIngoing(list);            
            }
        } catch (Exception e){
            e.printStackTrace();                
        }
    }
}

  public List<KpiMessage> fullPoll(int lastSeq) throws Exception {
    Statement st = dbConnection.createStatement();
    System.out.println("Polling");
 ResultSet rs = st.executeQuery("Select * from msg_new_to_bde where ACTION = 804 and SEQ >" + 
   lastSeq + "order by SEQ DESC");  

    return pojoCol;
}

运行方法处理：

     public void run() {

    try {
        generate(accumulator.outgoingQueue);
        accumulator.manageOutgoing(accumulator.outgoingQueue, dbConnection);
         } catch (Exception e) {
        e.printStackTrace();
    }
   }
  }

更新到数据库的方法

 public void updateDb(Collection<KpiMessage> updatedQueue, Connection dbConnection) throws  
  SQLException{ 
    for(KpiMessage pojoClass : updatedQueue){
            Statement stmtupd = dbConnection.createStatement();
        System.out.println("Updating");
    String query = "UPDATE msg_new_to_bde SET KEYINFO1= 'Processed', KEYINFO2 = 'Updated'
   WHERE ACTION = 804"; 

           stmtupd.executeUpdate(query);**My Execution stops here**

最后是一个用于维护所有这些队列的累加器类：

   public boolean isUsed = false;
    public synchronized void manageIngoing(List<KpiMessage> list){

    if(this.isUsed){                
        try {
            wait(); 
            System.out.println("first wait");
        } catch (Exception e1) {
            e1.printStackTrace();
        }
    }
    System.out.println("recived pass after update");
    this.getIncomingQueue().addAll(list);
     //incoming queue copied to outgoing queue
    this.setOutgoingQueue(this.getIncomingQueue());             
    System.out.println("waiting");
    System.out.println("new incoming message");
    this.isUsed = false;
    notifyAll();

}

/**
 * Method which handles synchronization using wait and notify for outgoing messages after   
  polling
 * @param outgoingQueue
 * @param dbConnection 
 */

  public synchronized void manageOutgoing(Collection<KpiMessage> outgoingQueue, Connection 
dbConnection){
    if(!this.isUsed)
    {
        try {
            System.out.println("second wait");
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    this.isUsed = true;
        DBhandler dbhandler = new DBhandler();
    try {
        dbhandler.updateDb(getOutgoingQueue(), dbConnection);
    } catch (SQLException e) {
        e.printStackTrace();
    }
    notifyAll();
}
 }

我的任务和问题是：

1.控制器应该同时处理线程轮询器和处理器以及累加器处理传入和传出队列，最后在处理后馈入更新队列以更新数据库

2.我这里的班级只轮询一次，无法更新，执行停止在

3.我的wait(), notifyALL() 句柄在这里是否正确。

这里如何实现重复轮询和更新？

4

1 回答 1

3

很有可能，在这个有五个不同问题的复杂环境中，所有事情都没有完整的答案。在等待这些时，您应该阅读 java.util.concurrent 必须提供的内容，尤其是支持阻塞读取和写入的并发集合。仅当 JDK 类对您来说不够用时，才使用wait()and 。notify()

于 2013-02-15T10:49:36.290 回答