package com.ard.tianyu.netty;
|
|
import com.ard.tianyu.config.TyConfiguration;
|
import com.ard.tianyu.model.cache.TyTcpServerCache;
|
import com.ard.tianyu.netty.handler.ClientHandler;
|
import com.ard.tianyu.netty.handler.MessageProtocolDecoder;
|
import com.ard.tianyu.netty.handler.MessageProtocolEncoder;
|
import io.netty.bootstrap.Bootstrap;
|
import io.netty.channel.*;
|
import io.netty.channel.nio.NioEventLoopGroup;
|
import io.netty.channel.socket.SocketChannel;
|
import io.netty.channel.socket.nio.NioSocketChannel;
|
import io.netty.handler.timeout.IdleStateHandler;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.PostConstruct;
|
import javax.annotation.Resource;
|
import java.util.HashMap;
|
import java.util.Map;
|
import java.util.concurrent.TimeUnit;
|
|
@Component
|
@Slf4j
|
public class TyNettyTcpClient {
|
|
public static Map<String, TyBoxSession> channels = new HashMap<>();
|
|
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
|
|
@Resource
|
private TyConfiguration tyConfiguration;
|
@Resource
|
ClientHandler clientHandler;
|
|
/**
|
* 初始化Bootstrap
|
*/
|
private Bootstrap getBootstrap() {
|
Bootstrap bootstrap = new Bootstrap();
|
bootstrap.group(eventLoopGroup)
|
.channel(NioSocketChannel.class)
|
.option(ChannelOption.TCP_NODELAY, true)
|
.option(ChannelOption.SO_KEEPALIVE, true)
|
.handler(new ChannelInitializer<SocketChannel>() {
|
@Override
|
protected void initChannel(SocketChannel socketChannel) throws Exception {
|
ChannelPipeline pipeline = socketChannel.pipeline();
|
pipeline.addLast("heart", new IdleStateHandler(0, 3, 0, TimeUnit.SECONDS));
|
pipeline.addLast("encoder", new MessageProtocolEncoder());
|
pipeline.addLast("decoder", new MessageProtocolDecoder());
|
pipeline.addLast("handler", clientHandler);
|
}
|
});
|
return bootstrap;
|
}
|
|
/**
|
* 初始化连接
|
*/
|
public void initConnection(String host, Integer port) {
|
|
log.info("初始化连接天宇tcp服务器【{}:{}】", host, port);
|
String currentBoxSn = "";
|
//判断是否已经存在连接，如果存在则关闭之前的连接
|
if (channels.containsKey(host + ":" + port)) {
|
currentBoxSn = channels.get(host + ":" + port).getSubBoxSn();
|
channels.get(host + ":" + port).getChannelFuture().channel().close();
|
}
|
|
//异步连接tcp服务端
|
Bootstrap bootstrap = getBootstrap();
|
ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> {
|
if (futureListener.isSuccess()) {
|
log.info("连接天宇TCP服务器成功【{}:{}】", host, port);
|
TyTcpServerCache.putTcpServerCache(host + ":" + port, true);
|
} else {
|
final EventLoop eventLoop = futureListener.channel().eventLoop();
|
eventLoop.schedule(() -> {
|
log.info("连接天宇TCP服务器失败【{}:{}】,正尝试重连...", host, port);
|
TyTcpServerCache.putTcpServerCache(host + ":" + port, false);
|
try {
|
initConnection(host, port);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}, 5, TimeUnit.SECONDS);
|
}
|
});
|
//将连接存入集合
|
channels.put(host + ":" + port, TyBoxSession.builder().subBoxSn(currentBoxSn).channelFuture(future).isReConnect(!currentBoxSn.equals("")).build());
|
}
|
|
//创建天宇连接
|
@PostConstruct
|
public void createTyConnect() {
|
try {
|
initConnection(tyConfiguration.getTcpserver(), tyConfiguration.getTcpport());
|
} catch (Exception e) {
|
log.error("创建TCP连接异常:{}", e.getMessage());
|
}
|
}
|
}
|
