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 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() { @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> portHosts) { for (Map 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()); } }