解决——并发请求丢失header请求头

港控/mmm° 2021-09-24 14:14 780阅读 0赞

前段时间在工作的过程中发现并发调用的情况下，部分请求会丢失请求头的部分信息，故在此做下记录

## 自定义Runnable ##

注意RequestAttributes这个字段，是解决问题的关键所在

@Slf4j
    public class TaskRunnable implements Runnable {
        private String taskName;
        private Runnable runnable;
        private Map<String, String> context;
        private RequestAttributes requestAttributes;
    
        public TaskRunnable(String name, Runnable runnable) {
            this.taskName = name;
            this.runnable = runnable;
            this.context = MDC.getCopyOfContextMap();
        }
    
        public TaskRunnable(String name, Runnable runnable, RequestAttributes requestAttributes) {
            this.taskName = name;
            this.runnable = runnable;
            this.context = MDC.getCopyOfContextMap();
            this.requestAttributes = requestAttributes;
        }
    
        @Override
        public void run() {
            try {
                if (this.context != null) {
                    MDC.setContextMap(context);
                }
                if (requestAttributes != null) {
                    RequestContextHolder.setRequestAttributes(requestAttributes);
                }
                runnable.run();
            } finally {
                if (requestAttributes != null) {
                    RequestContextHolder.resetRequestAttributes();
                }
                MDCUtils.clear();
            }
        }
    }

## 自定义callable ##

@Slf4j
    public class TaskCallable<T> implements Callable {
        private String taskName;
        private Callable<T> callable;
        private Map<String, String> context;
        private RequestAttributes requestAttributes;
    
        public TaskCallable(String name, Callable<T> callable) {
            this.taskName = name;
            this.callable = callable;
            this.context = MDC.getCopyOfContextMap();
        }
    
        public TaskCallable(String name, Callable<T> callable, RequestAttributes requestAttributes) {
            this.taskName = name;
            this.callable = callable;
            this.context = MDC.getCopyOfContextMap();
            this.requestAttributes = requestAttributes;
        }
    
        @Override
        public T call() throws Exception {
            try {
                if (this.context != null) {
                    MDC.setContextMap(context);
                }
                if (requestAttributes != null) {
                    RequestContextHolder.setRequestAttributes(requestAttributes);
                }
                return callable.call();
            } finally {
                if (requestAttributes != null) {
                    RequestContextHolder.resetRequestAttributes();
                }
                MDCUtils.clear();
            }
        }
    }

## 创建线程池 ##

`重点：RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();`

public class TaskThreadPoolExecutor extends ThreadPoolExecutor {
    
        private static final String DEFAULT_TASK_NAME = "Default Task";
    
        public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }
    
        public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        }
    
        public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        }
    
        public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                      RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        }
    
        @Override
        public void execute(Runnable command) {
            if (command instanceof TaskRunnable) {
                super.execute(command);
            } else {
                RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
                TaskRunnable taskCommand = new TaskRunnable(DEFAULT_TASK_NAME, command, requestAttributes);
                super.execute(taskCommand);
            }
        }
    }

## 使用方法 ##

private static final ExecutorService executorService = new TaskThreadPoolExecutor(20, 200,
                200L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(5000), new ThreadFactory() {
            final ThreadFactory defaultFactory = Executors.defaultThreadFactory();
    
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = defaultFactory.newThread(r);
                thread.setName("ShopTireService-" + thread.getName());
                return thread;
            }
        }, new ThreadPoolExecutor.AbortPolicy());

CompletableFuture<SearchTireProductResponse> future1 = CompletableFuture.supplyAsync(() -> getXXXService(), executorService);