사용자 지정 LinkedBlockingQueue 교착 상태

Question

ThreadExecutorPool 안에 사용자 지정 blockingqueue를 사용하고 있지만 작업 근로자가 작업을 수행하지 않고 dispissher 스레드가 새 작업을 큐에 넣지 않는 경우가 있습니다.

다음 사용자 지정 차단 대기열 구현으로 인해 교착 상태가 발생합니다. 이 코드에 문제가 있습니까? add() 및 take() 방법에 대해서는 synchronized 블록이 더 낫습니다.

import java.util.Collection; 
import java.util.concurrent.LinkedBlockingQueue; 

import org.apache.log4j.Logger; 

import com.ttech.utils.alarm.Alarm; 
import com.ttech.utils.alarm.AlarmInterface; 
import com.ttech.utils.counter.Counter; 
import com.ttech.utils.counter.SNMPAgent; 

public class WorkerQueue<E> extends LinkedBlockingQueue<E> { 

    private static final long serialVersionUID = 1L; 

    public Integer lowThreshold; 

    public Integer highThreshold; 

    public Integer capacity; 

    public String name; 

    public String type; 

    public Counter counter = null; 

    public boolean writeAlarmLog; 

    public static final Logger logger = Logger.getLogger(WorkerQueue.class); 

    public static Alarm HighThresholdAlarm = null; 
    public static Alarm CapacityAlarm = null; 

    // Check the size here and clear capacity and high threshold alarms in case 
    public E take() throws InterruptedException { 
     E data = super.take(); 
     counter.setNewValue(super.size()); 
     if (super.size() == lowThreshold) {    
      if(!this.writeAlarmLog) { 
       HighThresholdAlarm.clear(name); 
       CapacityAlarm.clear(name); 
      } else { 
       HighThresholdAlarm.clearLog(name, "Queue High Threshold"); 
       CapacityAlarm.clearLog(name, "Queue Capacity Overload"); 
      } 
     } 
     return data; 
    } 

    public E poll() { 
     E data = super.poll(); 
     counter.setNewValue(super.size()); 
     if (super.size() == lowThreshold) { 
      if(!this.writeAlarmLog) { 
       HighThresholdAlarm.clear(name); 
       CapacityAlarm.clear(name); 
      } else { 
       HighThresholdAlarm.clearLog(name, "Queue High Threshold"); 
       CapacityAlarm.clearLog(name, "Queue Capacity Overload"); 
      } 
     } 
     return data; 
    } 


    public int drainTo(Collection<? super E> c, int maxElements){ 
     int size = super.drainTo(c,maxElements);  
     counter.setNewValue(super.size());  
     return size; 
    } 

    // During adding the data to queue check capacity and high threshold raise alarm in case 
    public boolean add(E data) { 
     Boolean rc = true; 

     if (capacity > 0) { 
      if (this.size() >= capacity) { 
       logger.error("Queue " + name + " is over capacity"); 
       if(!this.writeAlarmLog) 
        CapacityAlarm.raise(name); 
       else 
        CapacityAlarm.raiseLog(AlarmInterface.AS_CRITICAL, name, "Queue Capacity Overload"); 
       return false; 
      } 
     } 

     if (!super.add(data)) { 
      logger.error("Cannot add data to queue:" + name); 
      rc = false; 
     } else { 
      counter.setNewValue(super.size()); 
     } 

     if (highThreshold == super.size()) { 


      if(!this.writeAlarmLog) 
       HighThresholdAlarm.raise(name); 
      else 
       HighThresholdAlarm.raiseLog(AlarmInterface.AS_CRITICAL, name, "Queue High Threshold"); 
     } 

     return rc; 
    } 
} 

Answer 1

ThreadPoolExecutor는 작업 큐에하지 add 작업을 수행합니다. 그것은 offers이고 허용되지 않으면 구성된 RejectedExecutionHandler으로 전달됩니다. 기본적으로이 값은 abort policy handler이며 RejectedExecutionException이 발생합니다.

사용자 지정 큐의 add 메서드는 절대로 호출되지 않습니다.

기내 작업 수의 변경 사항을 추적하려면 실행 프로그램 자체의 beforeExecute 또는 afterExecute 메서드를 재정의하는 것이 좋습니다. 활성 작업 수는 getActiveCount에서 얻을 수 있습니다.