diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java index 2c25614a..c3b8b439 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java @@ -353,4 +353,22 @@ public interface WxCpService { * @param wxConfigProvider */ public void setWxCpConfigStorage(WxCpConfigStorage wxConfigProvider); + + /** + *
+   * 设置当微信系统响应系统繁忙时,要等待多少 retrySleepMillis(ms) * 2^(重试次数 - 1) 再发起重试
+   * 默认:1000ms
+   * 
+ * @param retrySleepMillis + */ + void setRetrySleepMillis(int retrySleepMillis); + + /** + *
+   * 设置当微信系统响应系统繁忙时,最大重试次数
+   * 默认:5次
+   * 
+ * @param maxRetryTimes + */ + void setMaxRetryTimes(int maxRetryTimes); } diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java index ac13edaa..70ac819a 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java @@ -52,12 +52,14 @@ public class WxCpServiceImpl implements WxCpService { protected WxCpConfigStorage wxCpConfigStorage; - protected final ThreadLocal retryTimes = new ThreadLocal(); - protected CloseableHttpClient httpClient; protected HttpHost httpProxy; + private int retrySleepMillis = 1000; + + private int maxRetryTimes = 5; + public boolean checkSignature(String msgSignature, String timestamp, String nonce, String data) { try { return SHA1.gen(wxCpConfigStorage.getToken(), timestamp, nonce, data).equals(msgSignature); @@ -366,6 +368,33 @@ public class WxCpServiceImpl implements WxCpService { * @throws WxErrorException */ public T execute(RequestExecutor executor, String uri, E data) throws WxErrorException { + int retryTimes = 0; + do { + try { + return executeInternal(executor, uri, data); + } catch (WxErrorException e) { + WxError error = e.getError(); + /** + * -1 系统繁忙, 1000ms后重试 + */ + if (error.getErrorCode() == -1) { + int sleepMillis = retrySleepMillis * (1 << retryTimes); + try { + System.out.println("微信系统繁忙," + sleepMillis + "ms后重试(第" + (retryTimes + 1) + "次)"); + Thread.sleep(sleepMillis); + } catch (InterruptedException e1) { + throw new RuntimeException(e1); + } + } else { + throw e; + } + } + } while(++retryTimes < maxRetryTimes); + + throw new RuntimeException("微信服务端异常,超出重试次数"); + } + + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { String accessToken = getAccessToken(false); String uriWithAccessToken = uri; @@ -381,31 +410,10 @@ public class WxCpServiceImpl implements WxCpService { * 42001 access_token超时 */ if (error.getErrorCode() == 42001 || error.getErrorCode() == 40001) { - // 强制设置wxCpConfigStorage它的access token过期了,这样在下一次请求里就会刷新access token + // 强制设置wxMpConfigStorage它的access token过期了,这样在下一次请求里就会刷新access token wxCpConfigStorage.expireAccessToken(); return execute(executor, uri, data); } - /** - * -1 系统繁忙, 1000ms后重试 - */ - if (error.getErrorCode() == -1) { - if (retryTimes.get() == null) { - retryTimes.set(0); - } - if (retryTimes.get() > 4) { - retryTimes.set(0); - throw new RuntimeException("微信服务端异常,超出重试次数"); - } - int sleepMillis = 1000 * (1 << retryTimes.get()); - try { - System.out.println("微信系统繁忙," + sleepMillis + "ms后重试"); - Thread.sleep(sleepMillis); - retryTimes.set(retryTimes.get() + 1); - return execute(executor, uri, data); - } catch (InterruptedException e1) { - throw new RuntimeException(e1); - } - } if (error.getErrorCode() != 0) { throw new WxErrorException(error); } @@ -416,7 +424,6 @@ public class WxCpServiceImpl implements WxCpService { throw new RuntimeException(e); } } - protected CloseableHttpClient getHttpclient() { return httpClient; } @@ -451,4 +458,15 @@ public class WxCpServiceImpl implements WxCpService { } } + @Override + public void setRetrySleepMillis(int retrySleepMillis) { + this.retrySleepMillis = retrySleepMillis; + } + + + @Override + public void setMaxRetryTimes(int maxRetryTimes) { + this.maxRetryTimes = maxRetryTimes; + } + } diff --git a/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java new file mode 100644 index 00000000..cf129088 --- /dev/null +++ b/weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java @@ -0,0 +1,66 @@ +package me.chanjar.weixin.cp.api; + +import me.chanjar.weixin.common.bean.result.WxError; +import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.util.http.RequestExecutor; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +@Test +public class WxCpBusyRetryTest { + + @DataProvider(name="getService") + public Object[][] getService() { + WxCpService service = new WxCpServiceImpl() { + + @Override + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { + WxError error = new WxError(); + error.setErrorCode(-1); + throw new WxErrorException(error); + } + }; + + service.setMaxRetryTimes(3); + service.setRetrySleepMillis(500); + return new Object[][] { + new Object[] { service } + }; + } + + @Test(dataProvider = "getService", expectedExceptions = RuntimeException.class) + public void testRetry(WxCpService service) throws WxErrorException { + service.execute(null, null, null); + } + + @Test(dataProvider = "getService") + public void testRetryInThreadPool(final WxCpService service) throws InterruptedException, ExecutionException { + // 当线程池中的线程复用的时候,还是能保证相同的重试次数 + ExecutorService executorService = Executors.newFixedThreadPool(1); + Runnable runnable = new Runnable() { + @Override + public void run() { + try { + System.out.println("====================="); + System.out.println(Thread.currentThread().getName() + ": testRetry"); + service.execute(null, null, null); + } catch (WxErrorException e) { + throw new RuntimeException(e); + } catch (RuntimeException e) { + // OK + } + } + }; + Future submit1 = executorService.submit(runnable); + Future submit2 = executorService.submit(runnable); + + submit1.get(); + submit2.get(); + } + +} diff --git a/weixin-java-cp/src/test/resources/testng.xml b/weixin-java-cp/src/test/resources/testng.xml index 4cddddcf..ba92bc0c 100644 --- a/weixin-java-cp/src/test/resources/testng.xml +++ b/weixin-java-cp/src/test/resources/testng.xml @@ -1,28 +1,29 @@ - - - - - - - - - - - - + + + + + + + + + + + + + - - - - - - - - - - - + + + + + + + + + + + diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java index 7b194ccd..3ad10bdc 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java @@ -461,4 +461,22 @@ public interface WxMpService { * @param wxConfigProvider */ public void setWxMpConfigStorage(WxMpConfigStorage wxConfigProvider); + + /** + *
+   * 设置当微信系统响应系统繁忙时,要等待多少 retrySleepMillis(ms) * 2^(重试次数 - 1) 再发起重试
+   * 默认:1000ms
+   * 
+ * @param retrySleepMillis + */ + void setRetrySleepMillis(int retrySleepMillis); + + /** + *
+   * 设置当微信系统响应系统繁忙时,最大重试次数
+   * 默认:5次
+   * 
+ * @param maxRetryTimes + */ + void setMaxRetryTimes(int maxRetryTimes); } diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java index ba256865..744f8bf2 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java @@ -54,12 +54,14 @@ public class WxMpServiceImpl implements WxMpService { protected WxMpConfigStorage wxMpConfigStorage; - protected final ThreadLocal retryTimes = new ThreadLocal(); - protected CloseableHttpClient httpClient; protected HttpHost httpProxy; + private int retrySleepMillis = 1000; + + private int maxRetryTimes = 5; + public boolean checkSignature(String timestamp, String nonce, String signature) { try { return SHA1.gen(wxMpConfigStorage.getToken(), timestamp, nonce).equals(signature); @@ -439,7 +441,7 @@ public class WxMpServiceImpl implements WxMpService { return execute(new SimplePostRequestExecutor(), url, postData); } - /** + /** * 向微信端发送请求,在这里执行的策略是当发生access_token过期时才去刷新,然后重新执行请求,而不是全局定时请求 * @param executor * @param uri @@ -448,6 +450,33 @@ public class WxMpServiceImpl implements WxMpService { * @throws WxErrorException */ public T execute(RequestExecutor executor, String uri, E data) throws WxErrorException { + int retryTimes = 0; + do { + try { + return executeInternal(executor, uri, data); + } catch (WxErrorException e) { + WxError error = e.getError(); + /** + * -1 系统繁忙, 1000ms后重试 + */ + if (error.getErrorCode() == -1) { + int sleepMillis = retrySleepMillis * (1 << retryTimes); + try { + System.out.println("微信系统繁忙," + sleepMillis + "ms后重试(第" + (retryTimes + 1) + "次)"); + Thread.sleep(sleepMillis); + } catch (InterruptedException e1) { + throw new RuntimeException(e1); + } + } else { + throw e; + } + } + } while(++retryTimes < maxRetryTimes); + + throw new RuntimeException("微信服务端异常,超出重试次数"); + } + + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { String accessToken = getAccessToken(false); String uriWithAccessToken = uri; @@ -467,27 +496,6 @@ public class WxMpServiceImpl implements WxMpService { wxMpConfigStorage.expireAccessToken(); return execute(executor, uri, data); } - /** - * -1 系统繁忙, 1000ms后重试 - */ - if (error.getErrorCode() == -1) { - if(retryTimes.get() == null) { - retryTimes.set(0); - } - if (retryTimes.get() > 4) { - retryTimes.set(0); - throw new RuntimeException("微信服务端异常,超出重试次数"); - } - int sleepMillis = 1000 * (1 << retryTimes.get()); - try { - System.out.println("微信系统繁忙," + sleepMillis + "ms后重试"); - Thread.sleep(sleepMillis); - retryTimes.set(retryTimes.get() + 1); - return execute(executor, uri, data); - } catch (InterruptedException e1) { - throw new RuntimeException(e1); - } - } if (error.getErrorCode() != 0) { throw new WxErrorException(error); } @@ -533,4 +541,16 @@ public class WxMpServiceImpl implements WxMpService { } } + + @Override + public void setRetrySleepMillis(int retrySleepMillis) { + this.retrySleepMillis = retrySleepMillis; + } + + + @Override + public void setMaxRetryTimes(int maxRetryTimes) { + this.maxRetryTimes = maxRetryTimes; + } + } diff --git a/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java new file mode 100644 index 00000000..2e16723a --- /dev/null +++ b/weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java @@ -0,0 +1,66 @@ +package me.chanjar.weixin.mp.api; + +import me.chanjar.weixin.common.bean.result.WxError; +import me.chanjar.weixin.common.exception.WxErrorException; +import me.chanjar.weixin.common.util.http.RequestExecutor; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +@Test +public class WxMpBusyRetryTest { + + @DataProvider(name="getService") + public Object[][] getService() { + WxMpService service = new WxMpServiceImpl() { + + @Override + protected T executeInternal(RequestExecutor executor, String uri, E data) throws WxErrorException { + WxError error = new WxError(); + error.setErrorCode(-1); + throw new WxErrorException(error); + } + }; + + service.setMaxRetryTimes(3); + service.setRetrySleepMillis(500); + return new Object[][] { + new Object[] { service } + }; + } + + @Test(dataProvider = "getService", expectedExceptions = RuntimeException.class) + public void testRetry(WxMpService service) throws WxErrorException { + service.execute(null, null, null); + } + + @Test(dataProvider = "getService") + public void testRetryInThreadPool(final WxMpService service) throws InterruptedException, ExecutionException { + // 当线程池中的线程复用的时候,还是能保证相同的重试次数 + ExecutorService executorService = Executors.newFixedThreadPool(1); + Runnable runnable = new Runnable() { + @Override + public void run() { + try { + System.out.println("====================="); + System.out.println(Thread.currentThread().getName() + ": testRetry"); + service.execute(null, null, null); + } catch (WxErrorException e) { + throw new RuntimeException(e); + } catch (RuntimeException e) { + // OK + } + } + }; + Future submit1 = executorService.submit(runnable); + Future submit2 = executorService.submit(runnable); + + submit1.get(); + submit2.get(); + } + +} diff --git a/weixin-java-mp/src/test/resources/testng.xml b/weixin-java-mp/src/test/resources/testng.xml index 333400e4..cf9c2d64 100644 --- a/weixin-java-mp/src/test/resources/testng.xml +++ b/weixin-java-mp/src/test/resources/testng.xml @@ -3,6 +3,7 @@ +