Java多线程处理并汇总处理结果

迷南。 2024-02-17 09:56 24阅读 0赞

#### Java多线程处理并汇总处理结果 ####

*  简介
 *   *  1、示例代码
     *  2、线程池设置

## 简介 ##

在业务逻辑处理时，经常会for循环处理多条数据（如2000条），当每一条数据处理时间略微长时（0.1s），整体时间就会比较长了（200s）  
如果开10个线程处理时，整体时间就可以缩短10倍，整体只需要20s就可以完成了  
因此，本篇文章就介绍一下如何更好的多线程处理，以缩短处理时间

### 1、示例代码 ###

package ins.claim.simplecase.service;
    
    import com.google.common.collect.Lists;
    import com.google.common.util.concurrent.ListeningExecutorService;
    import ins.platform.common.util.CompletableFutureUtils;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.CompletableFuture;
    
    @Slf4j
    @Service
    public class TestThreadService {
        
    
        @Autowired
        private ListeningExecutorService threadPool;
    
    
        /**
         * @Author Author 
         * @Description 测试入口
         * @Date 16:58 2023/11/24
         * @Param []
         * @return java.util.List<java.util.Map<java.lang.String,java.lang.String>>
         **/
        public List<Map<String, String>> test(){
        
            // 1、组装请求数据
            List<Map<String, String>> dataList = new ArrayList<>();
            for (int i = 0; i < 1000; i++) {
        
                Map<String, String> map = new HashMap<>();
                map.put("1", "0");
                dataList.add(map);
            }
            // 2、多线程处理数据，并返回
            return this.deal(dataList);
        }
    
        /**
         * @Author Author 
         * @Description 分开多次线程处理
         * @Date 16:50 2023/11/24
         * @Param [list]
         * @return java.util.List<java.util.Map<java.lang.String,java.lang.String>>
         **/
        private List<Map<String, String>> deal(List<Map<String, String>> list){
        
            // 处理后返回的结果数据
            List<Map<String, String>> result = new ArrayList<>();
            // 每个线程处理的数据条数（可以自定义）
            int threadSize = 50;
            // 判断是否要开线程，计算需要开的线程个数，并依次处理
            if (list != null && list.size() > threadSize)
            {
        
                List<CompletableFuture<List<Map<String, String>>>> threadResultList = Lists.newArrayList();
                // 计算开几个线程，每个线程处理 threadSize 条
                int threadCount = (list.size() / threadSize) + 1;
                // 循环需要开的线程个数，并处理
                for (int i = 0; i < threadCount; i++) {
        
                    String rid = "日志信息" + i;
                    log.info("多线程处理第{}个线程", i);
                    if (i == threadCount - 1)
                    {
        
                        // 最后一次 开线程 处理
                        List<Map<String, String>> subList = list.subList(threadSize * i, list.size());
                        threadResultList.add(CompletableFuture.supplyAsync(() -> threadDealOnce(subList, rid), threadPool).exceptionally(ex -> {
        
                            log.error("线程处理失败了，请查看原因：", ex);
                            return null;
                        }));
                    }
                    else
                    {
        
                        // 开单个线程
                        List<Map<String, String>> subList = list.subList(threadSize * i, threadSize * (i + 1));
                        threadResultList.add(CompletableFuture.supplyAsync(() -> threadDealOnce(subList, rid), threadPool).exceptionally(ex -> {
        
                            log.error("线程处理失败了，请查看原因：", ex);
                            return null;
                        }));
                    }
                }
    
                // 多个线程处理后，结果合并处理
                for (List<Map<String, String>> resultOnce : CompletableFutureUtils.futureList(threadResultList)) {
        
                    result.addAll(resultOnce);
                }
            }
            else
            {
        
                // 不开线程，一次性处理
                result = threadDealOnce(list, "");
            }
    
            return result;
        }
    
        /**
         * @Author Author 
         * @Description 单次线程处理的方法（真实的业务方法）
         * @Date 16:49 2023/11/24
         * @Param [list, rid]
         * @return java.util.List<java.util.Map<java.lang.String,java.lang.String>>
         **/
        private List<Map<String, String>> threadDealOnce(List<Map<String, String>> list, String rid){
        
            log.info("单次线程处理的方法日志开始{}", rid);
            if (list != null && list.size() > 0) {
        
                int i = 0;
                for (Map<String, String> map : list) {
        
                    log.info("单次线程处理的方法日志{}", rid + "-" + i);
                    map.put("2", rid + i);
                    i++;
                }
            }
    
            return list;
        }
    }

/**
     * @Description 多线程处理数据测试
     * @return
     * @throws ParseException
     */
    @RequestMapping("/testThreadDealData")
    public List<Map<String, String>> testThreadDealData() throws Exception {
        
    	List<Map<String, String>> result = testThreadService.test();
    	return result;
    
    }

### 2、线程池设置 ###

服务启动时，需配置好线程池

import com.google.common.util.concurrent.ListeningExecutorService;
    import com.google.common.util.concurrent.MoreExecutors;
    import org.apache.commons.lang3.concurrent.BasicThreadFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.concurrent.*;
    
    /**
     * 在于Spring启动时自动加载一个ExecutorService对象. 得到一个20线程的线程池
     */
    @Configuration
    public class ThreadPoolConfig {
        
    
        /**
         * 该线程池使用的时候，不允许手动关闭，生命周期与应用绑定
         */
        @Bean
        public ListeningExecutorService getThreadPool(){
        
            ThreadFactory basicThreadFactory = new BasicThreadFactory.Builder().namingPattern("basicThreadFactory-ThreadPoolConfig").build();
            ListeningExecutorService pool = new MdcDelegateThreadPoolExecutor(MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(20, basicThreadFactory)));
            return pool;
        }
    
    }

import com.google.common.util.concurrent.AbstractListeningExecutorService;
    import com.google.common.util.concurrent.ListenableFuture;
    import com.google.common.util.concurrent.ListeningExecutorService;
    import com.sinosoft.fragins.framework.constance.ClmConstants;
    import com.sinosoft.fragins.framework.utils.UUIDUtils;
    import org.slf4j.MDC;
    
    import java.util.List;
    import java.util.concurrent.Callable;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Author Author 
     * @Description MDC rid 线程委托类(目的：rid 区分一下 主线程和子线程)
     * @Date 16:00 2023/11/6
     * @Param 
     * @return 
     **/
    public class MdcDelegateThreadPoolExecutor extends AbstractListeningExecutorService {
        
    
        private ListeningExecutorService delegate;
    
    
        public MdcDelegateThreadPoolExecutor(ListeningExecutorService delegate) {
        
            this.delegate = delegate;
        }
    
        @Override
        public ListenableFuture<?> submit(Runnable task) {
        
            return delegate.submit(new MdcDelegateRunnable(task));
        }
    
        @Override
        public <T> ListenableFuture<T> submit(Callable<T> task) {
        
            return delegate.submit(new MDCDelegateCallable<>(task));
        }
    
        @Override
        public void shutdown() {
        
            delegate.shutdown();
        }
    
        @Override
        public List<Runnable> shutdownNow() {
        
            return delegate.shutdownNow();
        }
    
        @Override
        public boolean isShutdown() {
        
            return delegate.isShutdown();
        }
    
        @Override
        public boolean isTerminated() {
        
            return delegate.isTerminated();
        }
    
        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        
            return delegate.awaitTermination(timeout, unit);
        }
    
        @Override
        public void execute(Runnable command) {
        
            delegate.execute(command);
        }
    
    
        class MDCDelegateCallable<V> implements Callable<V> {
        
    
            private String mainThreadRid;
    
            private Callable<V> delegate;
    
            public MDCDelegateCallable(Callable<V> callable) {
        
                this.mainThreadRid = MDC.get(ClmConstants.TRACE_ID_KEY);
                this.delegate = callable;
            }
    
            @Override
            public V call() throws Exception {
        
                MDC.put(ClmConstants.TRACE_ID_KEY, mainThreadRid + "-" + UUIDUtils.get12UUID());
                V ret = delegate.call();
                MDC.remove(ClmConstants.TRACE_ID_KEY);
                return ret;
            }
        }
    
        class MdcDelegateRunnable implements Runnable {
        
    
            // 主线程rid
            private String mainThreadRid;
    
            private Runnable delegate;
    
            /**
             * @Author Author 
             * @Description 构造器
             * @Date 16:01 2023/11/6
             * @Param [runnable]
             * @return
             **/
            public MdcDelegateRunnable(Runnable runnable) {
        
                this.mainThreadRid = MDC.get(ClmConstants.TRACE_ID_KEY);
                this.delegate = runnable;
            }
    
            @Override
            public void run() {
        
                MDC.put(ClmConstants.TRACE_ID_KEY, mainThreadRid + "-" + UUIDUtils.get12UUID());
                delegate.run();
                MDC.remove(ClmConstants.TRACE_ID_KEY);
            }
        }
    }