java自定义LinkedBlockingQueue死锁

Question

我一直在ThreadExecutorPool内使用自定义blockingqueue，但有时任务工作者不接受任务，dispacher线程也不将新任务放入队列

我想知道以下自定义阻塞队列实现会导致死锁。这个代码有什么错误吗？ 对于add()和take()方法，Is最好使用和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获得