Java线程池在运行后的结果反查

Question

我有一个在payara 5中运行的java应用程序。

我需要汇集我的bean将使用的一些引擎对象（来自库）。创建引擎需要在单独的线程中完成。

为此我想出了我的EnginePool和我的EngineProducer。想法是EnginePool管理两个BlockingQueues。一个用于可用引擎，另一个用于bean使用的引擎，需要再次使用。 EnginePool应该只提供一次，因此它是一个单例。

@Singleton
@Startup
@LocalBean
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public class AbbyyEnginePool implements OcrEnginePool<IEngine> {
private static final Logger logger = LoggerFactory.getLogger(AbbyyEnginePool.class);

@Resource(lookup = "java:comp/DefaultManagedThreadFactory")
private ManagedThreadFactory threadFactory;

private static final int DEFAULT_ENGINE_COUNT = 3;
private BlockingQueue<EngineMetaInfo> availableEngines = new ArrayBlockingQueue<>(DEFAULT_ENGINE_COUNT);
private BlockingQueue<IEngine> enginesToRelease = new ArrayBlockingQueue<>(DEFAULT_ENGINE_COUNT);
private Map<IEngine, IEngine> proxiesMapping = new ConcurrentHashMap<>(DEFAULT_ENGINE_COUNT);
private int poolSize;

public AbbyyEnginePool() {
    this(DEFAULT_ENGINE_COUNT);
}

public AbbyyEnginePool(int poolSize) {
    this.poolSize = poolSize;
    availableEngines = new ArrayBlockingQueue<>(poolSize);
    enginesToRelease = new ArrayBlockingQueue<>(poolSize);
    proxiesMapping = new ConcurrentHashMap<>(poolSize);
}

void setThreadFactory(ManagedThreadFactory threadFactory) {
    this.threadFactory = threadFactory;
}

@PostConstruct
void init() {
    EngineProducer engineProducer = new EngineProducer(availableEngines, enginesToRelease, poolSize);       
    Thread engineProducerThread = threadFactory.newThread(engineProducer);

    engineProducerThread.setName("engineProducer");
    engineProducerThread.start();

}

@Override
public  IEngine get() throws EngineException {
    try {
        EngineMetaInfo engineMetaInfo = availableEngines.take();
        IEngine engineProxy = IEngine.UnmarshalInterface(engineMetaInfo.engineHandle);
        proxiesMapping.put(engineProxy, engineMetaInfo.engine);
        return engineProxy;
    } catch (InterruptedException e) {
        throw new EngineException("Could not retrieve engine", e);
    }
}

@Override
public void release(IEngine engineProxy) throws EngineException {
    if (engineProxy != null) {
        synchronized (proxiesMapping) {
            if (proxiesMapping.containsKey(engineProxy)) {
                try {
                    IEngine engine = proxiesMapping.remove(engineProxy);
                    enginesToRelease.put(engine);
                } catch (Exception e) {
                    throw new EngineException("Could not release engine proxy.");
                }
            } else {
                logger.warn("Engine proxy was not registered. Could not release proxy.");
            }
        }
    }
}

static class EngineMetaInfo {
    long engineHandle;
    IEngine engine;

     EngineMetaInfo(long engineHandle, IEngine engine) {
        this.engineHandle = engineHandle;
        this.engine = engine;
    }
}

}

EngineProducer看起来像这样：

public class EngineProducer implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(EngineProducer.class);
private static final String PROJECT_ID = "someId";

private final Integer MAX_ENGINE_COUNT;
private final BlockingQueue<AbbyyEnginePool.EngineMetaInfo> availableEngines;
private final BlockingQueue<IEngine> enginesToRelease;

private Boolean isRunning = Boolean.FALSE;
private List<EngineHolder> enginesHolder;

public EngineProducer(BlockingQueue<AbbyyEnginePool.EngineMetaInfo> availableEngines,
                      BlockingQueue<IEngine> enginesToRelease,
                      Integer maxEnginesCount) {
    this.availableEngines = availableEngines;
    this.enginesToRelease = enginesToRelease;
    this.MAX_ENGINE_COUNT = maxEnginesCount;
    this.enginesHolder = new ArrayList<>(MAX_ENGINE_COUNT);
}

private void initEngines() {
    synchronized (availableEngines) {
        if (availableEngines.size() == 0) {
            try {
                for (int i = 0; i < MAX_ENGINE_COUNT; i++) {
                    EngineHolder engineHolder = new EngineHolder(PROJECT_ID);
                    enginesHolder.add(engineHolder);
                    IEngine engine = engineHolder.getAndLockEngine();
                    long engineHandle = engine.MarshalInterface();

                    AbbyyEnginePool.EngineMetaInfo engineMetaInfo = new AbbyyEnginePool.EngineMetaInfo(engineHandle, engine);
                    availableEngines.put(engineMetaInfo);
                }

                logger.info("{} abbyy engines prepared for processing", MAX_ENGINE_COUNT);
                EnginePool.setInitialized(Boolean.TRUE);
                isRunning = Boolean.TRUE;
            } catch (Exception e) {
                logger.error("Could not instantiate engines.", e);
            }
        }
    }
}

@Override
public void run() {
    try {
        initEngines();

        while(isRunning) {
            IEngine engineProxyToRelease = enginesToRelease.take();
            releaseEngine(engineProxyToRelease);
        }

        availableEngines.clear();
        for(int i = 0; i < enginesHolder.size(); i++) {
            enginesHolder.get(i).unloadEngine();
        }

    } catch (Exception e) {
        logger.error("EngineProducer encounter a problem.", e);
    }
}

public void unloadEngines() {
    isRunning = Boolean.FALSE;
}

private void releaseEngine( IEngine engineToRelease ) {
    for (EngineHolder engineHolder : enginesHolder) {
        if (engineHolder.containsEngine(engineToRelease)) {
            engineHolder.unlockEngine();

            IEngine engine = engineHolder.getAndLockEngine();
            long engineHandle = engine.MarshalInterface();
            AbbyyEnginePool.EngineMetaInfo engineMetaInfo = new AbbyyEnginePool.EngineMetaInfo(engineHandle, engine);
            try {
                availableEngines.put(engineMetaInfo);
            } catch (InterruptedException e) {
                logger.warn("could not add free engine");
            }
            break;
        }
    }
}

}

当我在测试中运行它而不是在glassfish中运行它时没有问题。但是当我在玻璃鱼中运行时，豆子会陷入僵局。

bean使用此代码来获取和释放引擎：

        engine = enginePool.get();
    ProcessingResult processingResult = null;
    try {
        this.parameters = parameters;

        this.tmpDir = tmpDir;
        Path customProfileFile = loadProfiles(parameters);

        Instant processingStart = Instant.now();
        processingResult = processFile();
        Instant processingEnd = Instant.now();
        enginePool.release(engine);
        engine = null;

        processingResult.setProcessingStartTime(processingStart);
        processingResult.setProcessingEndTime(processingEnd);

        logger.info("Processing took about {} milliseconds.", processingResult.getProcessDurationInMilliseconds());
        customProfileFile.toFile().delete();
        this.tmpDir.toFile().delete();
    } catch (Exception e) {
        logger.error("Ocr of document failed ",e );
        enginePool.release(engine);
        throw new EngineException("Ocr of document failed.", e);
    }

在我的场景中有4个尝试获取引擎的bean。其中3个将获得一个，最后一个bean将等待engine = enginePool.get(); 3个获得引擎的bean将完成他们的工作并等待enginePool.release(engine);。我接受了一个线程转储，可以看到3个bean正在等待一个没有引擎的bean所持有的锁。所以他们无法释放引擎。

我的问题是我不明白。释放和获取引擎在不同的阻塞队列上工作，所以我想知道为什么等待获取引擎的最后一个bean阻止试图释放引擎的其他bean。

Java线程池在运行后的结果反查

使用阻塞队列的对象池在死锁中运行