fix: 明确小店下载与路由背压行为

Author: binarywang

weixin-java-store/src/main/java/com/binarywang/wxjava/store/api/WxStoreBasicService.java

@@ -88,6 +88,7 @@ public interface WxStoreBasicService {
    * 根据media_id获取图片
    *
    * @param mediaId media_id
+   * @return 图片下载结果；调用方使用完 {@link StoreImageResponse#getFile()} 后必须删除该临时文件
    */
   StoreImageResponse getImg(String mediaId) throws WxErrorException;
 

weixin-java-store/src/main/java/com/binarywang/wxjava/store/executor/StoreMediaDownloadRequestExecutor.java

@@ -60,7 +60,6 @@ public static RequestExecutor<StoreImageResponse, String> create(RequestHttp<?,
   public static File createTmpFile(InputStream inputStream, String name, String ext, File tmpDirFile)
     throws IOException {
     File resultFile = File.createTempFile(name, '.' + ext, tmpDirFile);
-    resultFile.deleteOnExit();
     try (InputStream in = inputStream; OutputStream out = openOutputStream(resultFile)) {
       IOUtils.copy(in, out);
     }

weixin-java-store/src/main/java/com/binarywang/wxjava/store/message/WxStoreMessageRouter.java

@@ -35,7 +35,10 @@
 public class WxStoreMessageRouter {
   /** 规则列表 */
   private final List<WxStoreMessageRouterRule<? extends WxStoreMessage>> rules = new ArrayList<>();
-  /** 线程池 */
+  /**
+   * 线程池。默认使用容量为 1000 的有界队列；队列与最大线程数均耗尽时，
+   * 由提交消息的调用线程执行任务，以施加背压并避免丢弃回调消息。
+   */
   private ExecutorService executorService;
   /** 异常处理器 */
   private WxErrorExceptionHandler exceptionHandler;
@@ -47,7 +50,8 @@ public class WxStoreMessageRouter {
   public WxStoreMessageRouter() {
     ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("WxChMsgRouter-pool-%d").build();
     this.executorService = new ThreadPoolExecutor(2, 100,
-      0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), namedThreadFactory);
+      60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), namedThreadFactory,
+      new ThreadPoolExecutor.CallerRunsPolicy());
     this.sessionManager = new StandardSessionManager();
     this.exceptionHandler = new LogExceptionHandler();
     this.messageDuplicateChecker = WxMessageInMemoryDuplicateCheckerSingleton.getInstance();

weixin-java-store/src/test/java/com/binarywang/wxjava/store/message/WxStoreMessageRouterTest.java

@@ -0,0 +1,19 @@
+package com.binarywang.wxjava.store.message;
+
+import java.util.concurrent.ThreadPoolExecutor;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/** Tests the default router backpressure policy. */
+public class WxStoreMessageRouterTest {
+
+  @Test
+  public void shouldUseBoundedQueueAndCallerRunsPolicy() {
+    WxStoreMessageRouter router = new WxStoreMessageRouter();
+    ThreadPoolExecutor executor = (ThreadPoolExecutor) router.getExecutorService();
+
+    Assert.assertTrue(executor.getQueue().remainingCapacity() > 0);
+    Assert.assertTrue(executor.getRejectedExecutionHandler() instanceof ThreadPoolExecutor.CallerRunsPolicy);
+    router.shutDownExecutorService();
+  }
+}