package com.ard.tianyu.netty;
|
|
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.Resource;
|
import java.net.InetSocketAddress;
|
import java.util.HashMap;
|
import java.util.List;
|
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
|
ClientHandler clientHandler;
|
|
/**
|
* 初始化Bootstrap
|
*/
|
public Bootstrap getBootstrap(EventLoopGroup group) {
|
if (null == group) {
|
group = eventLoopGroup;
|
}
|
Bootstrap bootstrap = new Bootstrap();
|
bootstrap.group(group)
|
.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 initAllChannel(List<Map<String,Object>> portHosts) {
|
for (Map<String, Object> map : portHosts) {
|
String host = (String) map.get("IP");
|
Integer port = (Integer) map.get("PORT");
|
initOneConnection(host,port);
|
}
|
}
|
|
public void initOneConnection(String host,Integer port){
|
|
log.info("调用一次{}:{}",host,port);
|
//判断该账号下，是否有盒子被控制
|
//当前正在被控制的盒子
|
String currentBoxSn = "";
|
if(channels.containsKey(host+"/"+port))
|
{
|
currentBoxSn = channels.get(host+"/"+port).getSubBoxSn();
|
channels.get(host+"/"+port).getFuture().channel().close();
|
}
|
|
//异步连接tcp服务端
|
Bootstrap bootstrap = getBootstrap(null);
|
ChannelFuture future = bootstrap.connect(host,port).addListener((ChannelFuture futureListener) -> {
|
if (futureListener.isSuccess()) {
|
log.info("连接天宇TCP服务端成功:"+host+":"+port);
|
} else {
|
final EventLoop eventLoop = futureListener.channel().eventLoop();
|
eventLoop.schedule(new Runnable() {
|
@Override
|
public void run() {
|
log.info("连接天宇TCP服务端失败ip:{},port:{},正尝试重连...",host,port);
|
try {
|
initOneConnection(host,port);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
}, 5, TimeUnit.SECONDS);
|
}
|
});
|
|
//判断是否有盒子被订阅
|
channels.put(host+"/"+port, TyBoxSession.builder().subBoxSn(currentBoxSn).future(future)
|
.isReConnect(!currentBoxSn.equals("")).build());
|
}
|
}
|
