作者 crossoverJie

:bug: Fixing a bug.#70

... ... @@ -4,6 +4,7 @@ import com.crossoverjie.cim.client.config.AppConfiguration;
import com.crossoverjie.cim.client.init.CIMClientHandleInitializer;
import com.crossoverjie.cim.client.service.EchoService;
import com.crossoverjie.cim.client.service.MsgHandle;
import com.crossoverjie.cim.client.service.ReConnectManager;
import com.crossoverjie.cim.client.service.RouteRequest;
import com.crossoverjie.cim.client.service.impl.ClientInfo;
import com.crossoverjie.cim.client.thread.ContextHolder;
... ... @@ -67,6 +68,9 @@ public class CIMClient {
@Autowired
private ClientInfo clientInfo;
@Autowired
private ReConnectManager reConnectManager ;
/**
* 重试次数
*/
... ... @@ -110,7 +114,7 @@ public class CIMClient {
LOGGER.error("连接失败次数达到上限[{}]次", errorCount);
msgHandle.shutdown();
}
LOGGER.error("连接失败", e);
LOGGER.error("Connect fail!", e);
}
if (future.isSuccess()) {
echoService.echo("Start cim client success!");
... ... @@ -140,7 +144,7 @@ public class CIMClient {
errorCount++;
if (errorCount >= configuration.getErrorCount()) {
LOGGER.error("重连次数达到上限[{}]次", errorCount);
echoService.echo("The maximum number of reconnections has been reached[{}]times, close cim client!", errorCount);
msgHandle.shutdown();
}
LOGGER.error("login fail", e);
... ... @@ -198,6 +202,13 @@ public class CIMClient {
}
/**
* 1. clear route information.
* 2. reconnect.
* 3. shutdown reconnect job.
* 4. reset reconnect state.
* @throws Exception
*/
public void reconnect() throws Exception {
if (channel != null && channel.isActive()) {
return;
... ... @@ -207,7 +218,8 @@ public class CIMClient {
echoService.echo("cim server shutdown, reconnecting....");
start();
echoService.echo("Great! reconnect success!!!");
echoService.echo("Great! reConnect success!!!");
reConnectManager.reConnectSuccess();
ContextHolder.clear();
}
... ...
... ... @@ -87,7 +87,7 @@ public class BeanConfig {
@Bean("scheduledTask")
public ScheduledExecutorService buildSchedule(){
ThreadFactory sche = new ThreadFactoryBuilder()
.setNameFormat("reconnect-job-%d")
.setNameFormat("reConnect-job-%d")
.setDaemon(true)
.build();
ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,sche) ;
... ...
package com.crossoverjie.cim.client.handle;
import com.crossoverjie.cim.client.service.EchoService;
import com.crossoverjie.cim.client.service.ReConnectManager;
import com.crossoverjie.cim.client.service.ShutDownMsg;
import com.crossoverjie.cim.client.service.impl.EchoServiceImpl;
import com.crossoverjie.cim.client.thread.ReConnectJob;
import com.crossoverjie.cim.client.util.SpringBeanFactory;
import com.crossoverjie.cim.common.constant.Constants;
import com.crossoverjie.cim.common.protocol.CIMRequestProto;
... ... @@ -21,7 +21,6 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* Function:
... ... @@ -41,6 +40,8 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
private ScheduledExecutorService scheduledExecutorService ;
private ReConnectManager reConnectManager ;
private ShutDownMsg shutDownMsg ;
private EchoService echoService ;
... ... @@ -88,10 +89,11 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
if (scheduledExecutorService == null){
scheduledExecutorService = SpringBeanFactory.getBean("scheduledTask",ScheduledExecutorService.class) ;
reConnectManager = SpringBeanFactory.getBean(ReConnectManager.class) ;
}
LOGGER.info("客户端断开了,重新连接!");
// TODO: 2019-01-22 后期可以改为不用定时任务,连上后就关闭任务 节省性能。
scheduledExecutorService.scheduleAtFixedRate(new ReConnectJob(ctx),0,10, TimeUnit.SECONDS) ;
reConnectManager.reConnect(ctx);
}
@Override
... ...
package com.crossoverjie.cim.client.service;
import com.crossoverjie.cim.client.thread.ReConnectJob;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.channel.ChannelHandlerContext;
import org.springframework.stereotype.Component;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
/**
* Function:
*
* @author crossoverJie
* Date: 2020-04-15 00:26
* @since JDK 1.8
*/
@Component
public final class ReConnectManager {
private ScheduledExecutorService scheduledExecutorService;
/**
* Trigger reconnect job
* @param ctx
*/
public void reConnect(ChannelHandlerContext ctx) {
buildExecutor() ;
scheduledExecutorService.scheduleAtFixedRate(new ReConnectJob(ctx),0,10, TimeUnit.SECONDS) ;
}
/**
* Close reconnect job if reconnect success.
*/
public void reConnectSuccess(){
scheduledExecutorService.shutdown();
}
/***
* build an thread executor
* @return
*/
private ScheduledExecutorService buildExecutor() {
if (scheduledExecutorService == null || scheduledExecutorService.isShutdown()) {
ThreadFactory sche = new ThreadFactoryBuilder()
.setNameFormat("reConnect-job-%d")
.setDaemon(true)
.build();
scheduledExecutorService = new ScheduledThreadPoolExecutor(1, sche);
return scheduledExecutorService;
} else {
return scheduledExecutorService;
}
}
}
... ...
... ... @@ -143,7 +143,7 @@ public class RouteRequestImpl implements RouteRequest {
if (!cimServerResVO.getCode().equals(StatusEnum.SUCCESS.getCode())){
echoService.echo(cimServerResVO.getMessage());
// when client in reconnect state, could not exit.
// when client in reConnect state, could not exit.
if (ContextHolder.getReconnect()){
echoService.echo("###{}###", StatusEnum.RECONNECT_FAIL.getMessage());
throw new CIMException(StatusEnum.RECONNECT_FAIL);
... ...