자바 동시성 고수님들 도와주세요..

Question

안녕하세요 . 이제 1년차 자바 개발자입니다.threadpool을 공부하던 중. Threadpool을 만들어봤습니다.제가 의도했던건, 스레드 사이즈를 유동적으로 늘리고 싶었고, 풀사이즈가 꽉 차더라도 대기 큐에서 대기하도록 하고싶었습니다.일단 잘 동작하는 듯 했으나, 간헐적으로 작업 유실이 되는 문제가 있어서... 어디서 문제가 되는지 잘 모르겠습니다.고수님들 조언 부탁드리겠습니다.package org.example.thread.pool; import org.example.thread.policy.WaitingThreadExceptionPolicy; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; public final class DynamicThreadPool extends ThreadPoolExecutor { private final AtomicBoolean mainQueueFullFlag; public static ThreadPoolExecutor newDynamicThreadPool(int corePoolSize, int maximumPoolSize) { if (corePoolSize <= 0 || maximumPoolSize <= 0 || corePoolSize > maximumPoolSize) { throw new IllegalArgumentException(); } return new DynamicThreadPool(corePoolSize, maximumPoolSize); } public static ThreadPoolExecutor newDynamicThreadPool(int corePoolSize, BlockingQueue queue) { if (corePoolSize <= 0 || queue.remainingCapacity() <= 0 || corePoolSize > queue.remainingCapacity()) { throw new IllegalArgumentException(); } return new DynamicThreadPool(corePoolSize, queue.remainingCapacity(), queue); } private DynamicThreadPool(int corePoolSize, int maximumPoolSize) { super(corePoolSize, maximumPoolSize, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(maximumPoolSize), new WaitingThreadExceptionPolicy(new AtomicBoolean(false))); final WaitingThreadExceptionPolicy handler = (WaitingThreadExceptionPolicy) this.getRejectedExecutionHandler(); this.mainQueueFullFlag = handler.getMainQueueFullFlag(); } private DynamicThreadPool(int corePoolSize, int maximumPoolSize, BlockingQueue queue) { super(corePoolSize, maximumPoolSize, 60L, TimeUnit.SECONDS, queue, new WaitingThreadExceptionPolicy(new AtomicBoolean(false))); final WaitingThreadExceptionPolicy handler = (WaitingThreadExceptionPolicy) this.getRejectedExecutionHandler(); this.mainQueueFullFlag = handler.getMainQueueFullFlag(); } @Override public void execute(Runnable command) { if (this.mainQueueFullFlag.get()) { final WaitingThreadExceptionPolicy handler = (WaitingThreadExceptionPolicy) this.getRejectedExecutionHandler(); handler.getBlockingQueue().offer(command); return; } super.execute(command); } @Override protected void beforeExecute(Thread t, Runnable r) { final WaitingThreadExceptionPolicy handler = (WaitingThreadExceptionPolicy) this.getRejectedExecutionHandler(); synchronized (this.getQueue()) { final Runnable poll = handler.poll(); if (poll != null) { this.getQueue().offer(poll); } if (this.getQueue().remainingCapacity() > 0 && handler.getBlockingQueue().isEmpty()) { this.mainQueueFullFlag.compareAndSet(true, false); } } super.beforeExecute(t, r); } public Integer getWaitingQueueTaskCount() { final WaitingThreadExceptionPolicy waitingHandler = (WaitingThreadExceptionPolicy) this.getRejectedExecutionHandler(); return waitingHandler.getWaitingQueueCount(); } public static void main(String[] args) { ThreadPoolExecutor executor = DynamicThreadPool.newDynamicThreadPool(2, 10); for (int i = 0; i < 100; i++) { final int jobCnt = i; executor.submit(() -> { try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } System.out.println(Thread.currentThread().getName() + " ->>>>>> Running" + jobCnt); }); } int cnt = 0; while (true) { WaitingThreadExceptionPolicy rejectedExecutionHandler = (WaitingThreadExceptionPolicy) executor.getRejectedExecutionHandler(); System.out.println(rejectedExecutionHandler.getBlockingQueue().size()); System.out.println("Active Threads: " + executor.getActiveCount()); try { Thread.sleep(200); cnt += 1; if (cnt % 10 == 0){ for (int i = 0; i < 10; i++) { final int jobCnt = i; executor.submit(() -> { try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } System.out.println(Thread.currentThread().getName() + " ->>>>>> Running" + jobCnt); }); } } } catch (InterruptedException e) { throw new RuntimeException(e); } } } } -------------------------------------------------------------------------------- package org.example.thread.policy; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; public class WaitingThreadExceptionPolicy implements RejectedExecutionHandler { private final BlockingQueue blockingQueue; private final AtomicBoolean mainQueueFullFlag; public WaitingThreadExceptionPolicy(final AtomicBoolean mainQueueFullFlag){ this(new LinkedBlockingQueue<>(), mainQueueFullFlag); } public WaitingThreadExceptionPolicy(BlockingQueue blockingQueue, final AtomicBoolean mainQueueFullFlag) { this.blockingQueue = blockingQueue; this.mainQueueFullFlag = mainQueueFullFlag; } @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { if (!this.mainQueueFullFlag.get()) { this.mainQueueFullFlag.set(true); } boolean added = this.blockingQueue.offer(r); if (!added) { throw new RejectedExecutionException(); } } public BlockingQueue getBlockingQueue() { return blockingQueue; } public Runnable poll() { return this.blockingQueue.poll(); } public Integer getWaitingQueueCount() { return this.blockingQueue.size(); } public AtomicBoolean getMainQueueFullFlag() { return mainQueueFullFlag; } }

자바 동시성 고수님들 도와주세요..

0개의 답변

#threadpool 목록