Springboot - retry机制简介以及踩过的坑

像一般遇到这样的访问对端服务失败的情况我们都是怎么做的呢，一般不去主动处理的话，数据默认都丢弃了，对于一些对数据要求比较高的服务就不行了，要不就是去重试，要不就是在失败的时候将数据入库，等后面再人工介入处理。

org.springframework.web.client.ResourceAccessException: I/O error on POST request for "http://172.0.0.0:31113/api/device": Connect to 172.0.0.0:31113 [/172.0.0.0] failed: Connection refused (Connection refused); nested exception is org.apache.http.conn.HttpHostConnectException: Connect to 172.0.0.0:31113 [/172.0.0.0] failed: Connection refused (Connection refused)at org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:674)at org.springframework.web.client.RestTemplate.execute(RestTemplate.java:621)at org.springframework.web.client.RestTemplate.exchange(RestTemplate.java:539)at com.nbiot.server.adapter.util.StatefulRestTemplate.getEntity(StatefulRestTemplate.java:288)at com.nbiot.server.adapter.util.StatefulRestTemplate.postforEntity(StatefulRestTemplate.java:127)at com.nbiot.server.adapter.notify.EventManager.sendEnhancedAlarm(EventManager.java:430)at com.nokia.coap.adapter.notify.EventManager.access$400(EventManager.java:44)at com.nbiot.server.adapter.notify.EventManager$SendAlarmRunable.run(EventManager.java:216)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.http.conn.HttpHostConnectException: Connect to 172.0.0.0:31113 [/172.0.0.0] failed: Connection refused (Connection refused)at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:158)at org.apache.http.impl.conn.PoolingHttpClientConnectionManager.connect(PoolingHttpClientConnectionManager.java:353)at org.apache.http.impl.execchain.MainClientExec.establishRoute(MainClientExec.java:380)at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:236)at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:184)at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:88)at org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:184)at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:82)at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:55)at org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:89)at org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)at org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)at org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:660)... 10 common frames omitted
Caused by: java.net.ConnectException: Connection refused (Connection refused)at java.net.PlainSocketImpl.socketConnect(Native Method)at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)at java.net.Socket.connect(Socket.java:589)at org.apache.http.conn.socket.PlainConnectionSocketFactory.connectSocket(PlainConnectionSocketFactory.java:74)at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:141)... 23 common frames omitted

我们这里主要说一下重试机制的实现；

1、当然我们在不借助任何第三方工具和接口的情况下，直接用java的while循环，去不断的重试也是可以的，但是太过麻烦，而且还有重复‘造轮子’ 的嫌疑。。。

2、我们可以使用spring自带的retry机制；

①首先要在maven中加入spring-retry的依赖；

    <dependency><groupId>org.springframework.retry</groupId><artifactId>spring-retry</artifactId></dependency>

②启动类上添加 @EnableRetry

③在方法抛出异常需要重试的方法上面加上注解

backoff = @Backoff(value = 5000) ：重试延迟时间，单位毫秒，默认值1000，即默认延迟1秒

maxAttempts ：最大重试次数，默认为3。包括第一次失败

value ：需要进行重试的异常，和参数includes是一个意思。默认为空，当参数exclude也为空时，所有异常都将要求重试

@Retryable(value = RemoteAccessException.class, maxAttempts = 5, backoff = @Backoff(value = 5000))

④在重试次数达到最大的配置值时，进去@Recover标注的方法

该注解用于恢复处理方法，当全部尝试都失败时执行。返回参数必须和@Retryable修饰的方法返回参数完全一样。第一个参数必须是异常，其他参数和@Retryable修饰的方法参数顺序一致。

spring-retry 工具虽能优雅实现重试，但是经测试验证下来看，还有有所不足，对于现在项目来说不够灵活。。。

3 我们可以使用guava-retrying机制；

private boolean notify2Mark = true;
   private static final ConcurrentLinkedQueue<Notify2> notify2Queue = new ConcurrentLinkedQueue<>();
   private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("retry-pool-%d").build();
   private static final ExecutorService singleThreadPool = Executors.newSingleThreadExecutor(namedThreadFactory);
   //Compared with newSingleThreadExecutor ThreadPoolExecutor (corePoolSize = 1) runs services will bring substantial growth of MEM ?
   //private static ThreadPoolExecutor singleThreadPool = new ThreadPoolExecutor( 1, 1, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1), namedThreadFactory, new ThreadPoolExecutor.DiscardOldestPolicy());
   private static ThreadLocal<DateFormat> threadLocal = new ThreadLocal<DateFormat>() {
       @Override
       protected DateFormat initialValue() {
           return new SimpleDateFormat("yyyyMMdd HH:mm:ss,SSS");
       }
   };

... ...

try {
               if (notify2 != null && notify2Mark) {
                   sendEnhancedAlarm(notify2); //我们真正要重试的方法
               } else {
notify2Queue.offer(notify2); // 放入队列，保证可以在重试的时候拿到这条数据，且不影响后续数据的接收
logger.info("Queue busy, not been able to retry sending the message, temporarily stored in a queue, waiting to retry sending ... ");
}
           }catch (Exception e){
               logger.error("Thread end. " + this, e);
notify2Queue.offer(notify2);
notify2Mark = false;
               guavaRetry(); //发生异常，进入重试机制
           }

... ...

public void guavaRetry() {
final StringBuffer stringBuffer = new StringBuffer();
notify2 = notify2Queue.poll();
final Callable<Object> callable = new Callable<Object>() {
@Override
public Object call() throws Exception {
   if (notify2 != null){
                       logger.warn("Retry sending the message to Colocated Intelligent Gateway Server ... ");
                       sendEnhancedAlarm(notify2);//我们真正要重试的方法，并且在这里执行guava的重试逻辑；
                   }
return true;
}
};
           final Retryer<Object> retryer = RetryerBuilder.newBuilder()
                   .retryIfExceptionOfType(IOException.class)
                   .retryIfException(Predicates.or(
                           Predicates.instanceOf(ConnectException.class),
                           Predicates.instanceOf(ResourceAccessException.class),
                           Predicates.instanceOf(HttpHostConnectException.class),
                           Predicates.instanceOf(ConnectTimeoutException.class),
                           Predicates.instanceOf(SocketTimeoutException.class),
                           Predicates.instanceOf(RequestAbortedException.class)
                           )
                   )
                   //.withStopStrategy(StopStrategies.stopAfterAttempt(2)) //停止策略最大失败次数2次
                   .withStopStrategy(StopStrategies.stopAfterDelay(60, TimeUnit.SECONDS)) //停止策略最大失败时间60s
                   .withWaitStrategy(WaitStrategies.fixedWait(5, TimeUnit.SECONDS)) //每隔5s重试一次
                   //默认的阻塞策略：线程睡眠
                   .withBlockStrategy(BlockStrategies.threadSleepStrategy())
                   //自定义阻塞策略：自旋锁会造成cpu空等，而导致cpu占用过高
                   //.withBlockStrategy(new SpinBlockStrategy())
                   .withRetryListener(new RetryListener() {
                       @Override
                       public <V> void onRetry(final Attempt<V> attempt) {
                           logger.info(" [retry]time=[{}]", attempt.getAttemptNumber());
                           logger.warn(" hasException=[{}]", attempt.hasException());
                           logger.warn(" hasResult=[{}]", attempt.hasResult());
                           if (attempt.hasException()) {
                               logger.error(" causeBy=[{}]", attempt.getExceptionCause().toString());
                           } else {
                               logger.info(" retryResult=[{}] and retry successfully ", attempt.getResult());

                               if (!notify2Queue.isEmpty()) {
                                   singleThreadPool.execute(() -> { guavaRetry(); });
                               } else {
                                   notify2Mark = true;
                                   logger.info("Queue has been emptied, the new data into the waiting queue ... ");
                               }
                           }
                       }
                   }).build();

try {
retryer.call(callable);
} catch (RetryException | ExecutionException e) {
Object payloadValue = "", tagValue = "", endpointValue = "";
logger.warn("Will do something such as save message to file or db ; {}", e.getMessage());
for (int i = 0; i < notify2.getNotify2().size(); ++i) {
if ("uplinkMsg/0/data".equals(this.notify2.getNotify2().get(i).get("resource"))) {
payloadValue = notify2.getNotify2().get(i).get("value");
endpointValue = notify2.getNotify2().get(i).get("device/0/endPointClientName");
}
if ("device/0/tag".equals(notify2.getNotify2().get(i).get("resource"))) {
tagValue = notify2.getNotify2().get(i).get("value");
}
}

FileUtilManager.appendWriteFile(stringBuffer.append(threadLocal.get().format(new Date()))
.append(",").append(payloadValue).append(",").append(tagValue).append(",").append(endpointValue).toString());

if(!notify2Queue.isEmpty()){
singleThreadPool.execute(() -> { guavaRetry(); });
} else {
notify2Mark = true;
logger.info("Queue has been emptied, the new data into the waiting queue ... ");
}
}
}

... ...
/**
* 生成文件，并以追加的方式写入到文件中
*/
public static void appendWriteFile(String conent) {
File path = new File("retryfile");
if (!path.exists()) { path.mkdirs();}
String file = "retryfile/retryfile_" + threadLocal.get().format(new Date()) + ".payload";
logger.info("Start to write the file in appended form to :[" + file + "]");
BufferedWriter out = null;
try {
out = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file, true)));
out.write(conent + "\n");
logger.info("Write file :[" + file + "] complete");
} catch (Exception e) {
logger.error("Write file :[" + file + "] exception, the exception information is:[" + e.getMessage() + "]");
} finally {
logger.info("Start to close the output stream");
try {
out.close();
} catch (IOException e) {
logger.info("Close the output stream exception, the exception information is:[" + e.getMessage() + "]");
}
}
}

... ...

/*** 自旋锁的实现, 不响应线程中断*/public class SpinBlockStrategy implements BlockStrategy {@Overridepublic void block(final long sleepTime) throws InterruptedException {final LocalDateTime startTime = LocalDateTime.now();long end;final long start = end = System.currentTimeMillis();logger.info("[SpinBlockStrategy]...begin wait.");while (end - start <= sleepTime) {end = System.currentTimeMillis();}final Duration duration = Duration.between(startTime, LocalDateTime.now());logger.info("[SpinBlockStrategy]...end wait.duration={}", duration.toMillis());}}

使用guava的retry相对来说还是比较灵活的，可以添加RetryListener来实时监控重试的状态，可以使用retryIfException来自定义重试的异常，或者异常的类型，而且RetryerBuilder也是线程安全的。

。

Springboot - retry机制简介以及踩过的坑相关推荐

SpringBoot常用配置简介
SpringBoot常用配置简介 1. SpringBoot中几个常用的配置的简单介绍一个简单的Spring.factories # Bootstrap components org.springf ...
BlockChain：《Blockchain Gate》听课笔记——区块链的共识机制—简介、理解、畅谈
BlockChain:<Blockchain Gate>听课笔记--区块链的共识机制-简介.理解.畅谈以下资源为各种渠道的网络收集和个人总结目录区块链--比特币中的共识机制现阶段区 ...
Linux Namespace机制简介
最近Docker技术越来越受到关注,作为Docker中很重要的一项技术,Namespace也就经常在Docker的简介里面看到. 在这里总结一下它的内部机制.也解决一下自己原来的一些疑惑. Names ...
Springboot使用Filter以及踩过的坑
Springboot使用Filter以及踩过的坑在Springboot中使用Filter有两种方式,注解方式,注册bean方式一.注解@WebFilter 1.实现Filter接口(javax.s ...
tenacity发生异常/失败/错误时重试retry机制，Python
tenacity发生异常/失败/错误时重试retry机制,Python 安装: pip install tenacity 示例: @retry def non_stop():print("永 ...
【一篇就够了】springboot微信公众号开发，你的坑我来踩
[一篇就够了]springboot微信公众号开发,你的坑我来踩前些日子在抖音上看到一个写给女朋友的微信公众号突然心血来潮自己也想写一个,随后就开始在下面的踩坑填坑的阶段了,因为也是第一次写微信公众号 ...
TCP/UDP报文格式及各种通信机制简介
TCP/UDP报文格式及各种通信机制简介一.UDP报文二.TCP报文三.TCP通信机制 1,确认应答机制 2,超时重传机制 3,滑动窗口及快重传机制 4,流量控制 5,拥塞控制及慢启动机制 6, ...
2019秋招总结（非科班转互联网，请不要踩我的坑）
LZ本硕某985,专业是电子信息相关,秋招找的工作都是Java后端相关方向,投了有几十家,情况大致如下: 简历挂: 中电十四所(估计已经是招满了),陌陌(感觉是宣讲会直接放了简历就走了,所以简历没过) ...
java项目经理也就那么回事_网易PM | 我们之前在需求评审环节踩过的坑...
原本觉得需求评审也就那么回事儿,大家应该都差不多这么做的,没啥好说的.不过前不久有一位同学问起来我们是怎么做需求评审的,然后发现有一些团队的做法可能还不大一样,他们也还踩着我们之前踩过的坑,他们还在探 ...

Springboot - retry机制简介以及踩过的坑

Springboot - retry机制简介以及踩过的坑相关推荐

最新文章

热门文章