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("初始化连接天宇tcp服务器【{}:{}】", 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(() -> { log.info("连接天宇TCP服务器失败【{}:{}】,正尝试重连...", 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()); } }