有狀態InheritableThreadLocal 配合 JDK8 ,異步方法調用

我們可以把一個類的作用域注解為

@Scope(scopeName = WebApplicationContext.SCOPE_SESSION, proxyMode = ScopedProxyMode.TARGET_CLASS)

這樣這個類就能在session中獲取,可以把用戶信息放到這個類中,需要的時候,直接@Autowire進來就好了.

但是這樣有一個坑.在主線程中,如果使用JDK異步方法,或者自己new出新的線程中,沒有辦法注入.會提示一個異常

Scope 'session' is not active for the current thread; consider defining a scoped proxy for this bean if you intend to refer to it from a singleton;

這個異常非常誤導人,在單線程下是不會提示的.提示了這個異常,原因是:session 是根據request獲取的,那么Spring是用requestContextHolder實現的,原理還是InheritableThreadLocal ,按道理來說,在子線程也是可以獲取到父線程的request.getSession,至於為什么在子線程中沒有激活這個session作用域,導致注入失敗,估計是Spring的設計問題.

所以參考RequestContextHolder,自己可以寫一個記錄結果以及錯誤信息的LocalResut

 雖然我用了沒有問題,但是所有書籍記錄中,都不推薦在並發流或者多線程中使用有狀態對象,雖然我這不涉及競態條件,但是如果因為線程回收慢導致ThreadLocal數據沒有回寫到主線程就已經返回了,或者可見性問題,那么結果也會有誤,而且不好debug

這里用一個ThreadLocal 記錄結果集,或者錯誤信息,等調用完並發流以后,可以join之后,輸出結果,注意線程可見性問題,這加上volatile 修飾.

public class LocalResult{
    private static volatile ThreadLocal<Map<String, String>> mapMessage = new InheritableThreadLocal();

    public LocalError() {
    }


    public static void mapMessageCreate() {
        mapMessage.remove();
        mapMessage.set(new HashMap());
    }

    public static ThreadLocal<Map<String, String>> getMapMessage() {
        return mapMessage;
    }
}

 

在這里根據訂單的id數組,進行批量的異步結算,在結算方法中,如果有錯誤信息,依然會調用LocalResult,

線程池等於需要並發結算的數量+3 是一個容錯,避免線程池的Thead不夠,導致ThreadLocal數據亂套

for開啟異步方法,之后馬上關閉線程池

然后還要等待線程池所有線程已經回收以后才讓方法返回:是因為線程執行完方法,不一定立即回收,當線程已經開啟,馬上調用線程池的shutdown方法,告訴線程執行完后,盡快回收.

然后循環判斷線程池是否完全回收后,才返回因為,根據join()規則。假設線程A在執行的過程中，通過執行ThreadB.join()來等待線

程B終止；同時，假設線程B在終止之前修改了一些共享變量，線程A從ThreadB.join()返回后會
讀這些共享變量。

.

   public void doAsync(Integer [] ids) {
        List<CompletableFuture<Boolean>> futures = new ArrayList<>(ids.length);
        ExecutorService executor = Executors.newFixedThreadPool(ids.length+3);
        for (Integer id : ids) {
            Order order = orderService.findEntityById(id);
            if (Order.Type.Failure.getCode().equals(orderInfo.getType())) {
                LocalResult.getMapMessage().get().put(order.getOrderNo(), "單號:" + order.getOrderNo() + ",已作廢,不能結算!");
                continue;
            }
            futures.add(CompletableFuture.supplyAsync(() -> this.do(orderInfo),executor));
        }
        executor.shutdown();
        futures.stream().map(CompletableFuture::join);
        while (!executor.isTerminated()) {
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                break;
            }
        }
    }

 controller

    @PostMapping(path = "/do",produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public ResultDataDto billing(@RequestParam("ids") Integer[] ids) {
        LocalResult.mapMessageCreate();
        service.doAsync(ids);
        if (LocalResult.getMapMessage().get().size() > 0) {
            StringBuilder stringBuilder = new StringBuilder();
            for (Map.Entry<String, String> entry : LocalResult.getMapMessage().get().entrySet()) {
                stringBuilder.append("單號:" + entry.getKey() + ":" + entry.getValue() + "</br>");
            }
            return ResultDataDto.addOperationSuccess("以下失敗!:</br>" + stringBuilder.toString());
        }
        return ResultDataDto.addOperationSuccess("結算成功!");
    }