Procházet zdrojové kódy

fix: WxMpServiceImpl中用ThreadLocal记录重试次数,这个是有问题的,因为在多线程(线程池)环境下ThreadLocal是不会被清0的

master
Daniel Qian před 10 roky
rodič
revize
e53c921d17
8 změnil soubory, kde provedl 280 přidání a 72 odebrání
  1. +18
    -0
      weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java
  2. +43
    -25
      weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java
  3. +66
    -0
      weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java
  4. +24
    -23
      weixin-java-cp/src/test/resources/testng.xml
  5. +18
    -0
      weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java
  6. +44
    -24
      weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java
  7. +66
    -0
      weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java
  8. +1
    -0
      weixin-java-mp/src/test/resources/testng.xml

+ 18
- 0
weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpService.java Zobrazit soubor

@@ -353,4 +353,22 @@ public interface WxCpService {
* @param wxConfigProvider
*/
public void setWxCpConfigStorage(WxCpConfigStorage wxConfigProvider);

/**
* <pre>
* 设置当微信系统响应系统繁忙时,要等待多少 retrySleepMillis(ms) * 2^(重试次数 - 1) 再发起重试
* 默认:1000ms
* </pre>
* @param retrySleepMillis
*/
void setRetrySleepMillis(int retrySleepMillis);

/**
* <pre>
* 设置当微信系统响应系统繁忙时,最大重试次数
* 默认:5次
* </pre>
* @param maxRetryTimes
*/
void setMaxRetryTimes(int maxRetryTimes);
}

+ 43
- 25
weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpServiceImpl.java Zobrazit soubor

@@ -52,12 +52,14 @@ public class WxCpServiceImpl implements WxCpService {

protected WxCpConfigStorage wxCpConfigStorage;

protected final ThreadLocal<Integer> retryTimes = new ThreadLocal<Integer>();

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, E> T execute(RequestExecutor<T, E> 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, E> T executeInternal(RequestExecutor<T, E> 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;
}

}

+ 66
- 0
weixin-java-cp/src/test/java/me/chanjar/weixin/cp/api/WxCpBusyRetryTest.java Zobrazit soubor

@@ -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, E> T executeInternal(RequestExecutor<T, E> 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();
}

}

+ 24
- 23
weixin-java-cp/src/test/resources/testng.xml Zobrazit soubor

@@ -1,28 +1,29 @@
<!DOCTYPE suite SYSTEM "http://testng.org/testng-1.0.dtd" >

<suite name="Weixin-java-tool-suite" verbose="1">
<test name="API_Test">
<classes>
<class name="me.chanjar.weixin.cp.api.WxCpBaseAPITest" />
<class name="me.chanjar.weixin.cp.api.WxCpMessageAPITest" />
<class name="me.chanjar.weixin.cp.api.WxMenuAPITest" />
<class name="me.chanjar.weixin.cp.api.WxCpDepartAPITest" />
<class name="me.chanjar.weixin.cp.api.WxCpMediaAPITest" />
<class name="me.chanjar.weixin.cp.api.WxCpMessageRouterTest" />
<class name="me.chanjar.weixin.cp.api.WxCpTagAPITest" />
<class name="me.chanjar.weixin.cp.api.WxCpUserAPITest" />
</classes>
</test>
<test name="API_Test">
<classes>
<class name="me.chanjar.weixin.cp.api.WxCpBusyRetryTest"/>
<class name="me.chanjar.weixin.cp.api.WxCpBaseAPITest"/>
<class name="me.chanjar.weixin.cp.api.WxCpMessageAPITest"/>
<class name="me.chanjar.weixin.cp.api.WxMenuAPITest"/>
<class name="me.chanjar.weixin.cp.api.WxCpDepartAPITest"/>
<class name="me.chanjar.weixin.cp.api.WxCpMediaAPITest"/>
<class name="me.chanjar.weixin.cp.api.WxCpMessageRouterTest"/>
<class name="me.chanjar.weixin.cp.api.WxCpTagAPITest"/>
<class name="me.chanjar.weixin.cp.api.WxCpUserAPITest"/>
</classes>
</test>

<test name="Bean_Test">
<classes>
<class name="me.chanjar.weixin.cp.bean.WxCpMessageTest" />
<class name="me.chanjar.weixin.cp.bean.WxCpXmlMessageTest" />
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutImageMessageTest" />
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutNewsMessageTest" />
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutVideoMessageTest" />
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutVoiceMessageTest" />
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutTextMessageTest" />
</classes>
</test>
<test name="Bean_Test">
<classes>
<class name="me.chanjar.weixin.cp.bean.WxCpMessageTest"/>
<class name="me.chanjar.weixin.cp.bean.WxCpXmlMessageTest"/>
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutImageMessageTest"/>
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutNewsMessageTest"/>
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutVideoMessageTest"/>
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutVoiceMessageTest"/>
<class name="me.chanjar.weixin.cp.bean.WxCpXmlOutTextMessageTest"/>
</classes>
</test>
</suite>

+ 18
- 0
weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpService.java Zobrazit soubor

@@ -461,4 +461,22 @@ public interface WxMpService {
* @param wxConfigProvider
*/
public void setWxMpConfigStorage(WxMpConfigStorage wxConfigProvider);

/**
* <pre>
* 设置当微信系统响应系统繁忙时,要等待多少 retrySleepMillis(ms) * 2^(重试次数 - 1) 再发起重试
* 默认:1000ms
* </pre>
* @param retrySleepMillis
*/
void setRetrySleepMillis(int retrySleepMillis);

/**
* <pre>
* 设置当微信系统响应系统繁忙时,最大重试次数
* 默认:5次
* </pre>
* @param maxRetryTimes
*/
void setMaxRetryTimes(int maxRetryTimes);
}

+ 44
- 24
weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpServiceImpl.java Zobrazit soubor

@@ -54,12 +54,14 @@ public class WxMpServiceImpl implements WxMpService {

protected WxMpConfigStorage wxMpConfigStorage;
protected final ThreadLocal<Integer> retryTimes = new ThreadLocal<Integer>();

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, E> T execute(RequestExecutor<T, E> 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, E> T executeInternal(RequestExecutor<T, E> 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;
}

}

+ 66
- 0
weixin-java-mp/src/test/java/me/chanjar/weixin/mp/api/WxMpBusyRetryTest.java Zobrazit soubor

@@ -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, E> T executeInternal(RequestExecutor<T, E> 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();
}

}

+ 1
- 0
weixin-java-mp/src/test/resources/testng.xml Zobrazit soubor

@@ -3,6 +3,7 @@
<suite name="Weixin-java-tool-suite" verbose="1">
<test name="API_Test">
<classes>
<class name="me.chanjar.weixin.mp.api.WxMpBusyRetryTest" />
<class name="me.chanjar.weixin.mp.api.WxMpBaseAPITest" />
<class name="me.chanjar.weixin.mp.api.WxMpCustomMessageAPITest" />
<class name="me.chanjar.weixin.mp.api.WxMpMenuAPITest" />


Načítá se…
Zrušit
Uložit