java - 处理线程池并等待notifyALL()

Question

如何处理线程池，其中一个正在轮询，而另一个应在处理后更新新传入的数据。程序执行在 Controller 类中，该类具有 main 方法和线程池:

主类Controller

   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. Controller 应该处理线程轮询器和处理器，累加器处理传入和传出队列，最后输入到更新队列以在处理后更新数据库

2.我这里的类只进行一次轮询，无法更新，执行停止于

3.我的 wait()、notifyALL() 处理是否正确？

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

Answer 1

在这个包含五个不同问题的复杂环境中，很可能不会有所有问题的完整答案。在等待这些的同时，您应该阅读 java.util.concurrent 所提供的内容，尤其是支持阻塞读取和写入的并发集合。仅当 JDK 类不足以满足您的需求时，才使用 wait() 和 notify()。