自执行线程自启动处理队列线程池限流执行

ゝ一纸荒年。 2023-07-12 03:35 11阅读 0赞

import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.google.common.util.concurrent.*;
    import org.apache.commons.lang3.concurrent.BasicThreadFactory;
    
    import javax.annotation.Nullable;
    import java.util.Queue;
    import java.util.concurrent.*;
    
    /** * @author ： * @date ：Created in 2020/3/6 13:31 * @description：${description} */
    public class test implements Runnable { 
    
    
        public test() { 
            executorService.scheduleWithFixedDelay(this,10,20, TimeUnit.SECONDS);
        }
    
        private static ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1, Executors.defaultThreadFactory());
    
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10,10, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(5000),new BasicThreadFactory.Builder().build(),new ThreadPoolExecutor.AbortPolicy());
    
        ListeningExecutorService listeningDecorator = MoreExecutors.listeningDecorator(executor);
    
    
        private static final Queue<JSONObject> workQueue = new LinkedBlockingQueue<JSONObject>();
    
        private Semaphore available = null;
    
        @Override
        public void run() { 
    
            try { 
                if (available == null) { 
                    available = new Semaphore(10, true);
                }
                if( workQueue.size()>0 ) { 
                    JSONObject po = workQueue.poll();
                    while (null != po) { 
                        available.acquire();
                        ListenableFuture<JSONObject> test = listeningDecorator.submit(new Callable<JSONObject>() { 
                            @Override
                            public JSONObject call() throws Exception { 
                                JSONObject js = JSON.parseObject("TEST");
                                return js;
                            }
                        });
    
                        FutureCallback futureCallback = new FutureCallback() { 
                            @Override
                            public void onSuccess(@Nullable Object o) { 
                                available.release();
                            }
                            @Override
                            public void onFailure(Throwable throwable) { 
                                available.release();
                            }
                        };
                        Futures.addCallback(test,futureCallback,Executors.newSingleThreadExecutor());
                    }
                }
            }catch (Exception e){ 
            }
    
        }
    
    }