Netty客户端

TCPClient

public class NettyTcpClient {/*** TCP服务端IP(MD)*/@Value("${mdServer.ip}")private String ip;/*** TCP服务端端口号(MD)*/@Value("${mdServer.port}")private int port;@Resourceprivate UnvChannelInitialize channelInitialize;private Bootstrap bootstrap = new Bootstrap();private List<Channel> channelList = new ArrayList<>();private final AtomicInteger index = new AtomicInteger();/*** 初始化TCP客户端*/public void initClient() {int threadNum = Runtime.getRuntime().availableProcessors();if (threadNum <= 0) {threadNum = 4;}EventLoopGroup loopGroup = new NioEventLoopGroup();try {bootstrap.group(loopGroup).channel(NioSocketChannel.class).remoteAddress(new InetSocketAddress(ip, port)).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true)// 设置ChanelOutBoundBuffer缓冲区最低水位为64kb,最高水位为256kb.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(32 * 1024, 256 * 1024)).handler(channelInitialize);// 创建多个连接for (int i = 0; i < threadNum; i++) {ChannelFuture channelFuture = bootstrap.connect().sync();channelList.add(channelFuture.channel());}log.info("{} 服务端连接成功", ip + ":" + port);} catch (Exception e) {log.error("TCP客户端连接{}服务端失败:", ip + ":" + port, e);}}/*** 获取Chanel实例** @return channel*/public Channel getNextChannel() {return getFirstActiveChannel(0);}/*** 获取Chanel实例** @param count* @return channel*/public Channel getFirstActiveChannel(int count) {Channel channel = channelList.get(Math.abs(index.getAndIncrement() % channelList.size()));if (!channel.isActive()) {// 重连reConnect(channel);if (count > channelList.size()) {throw new RuntimeException("暂无空闲的连接通道");}return getFirstActiveChannel(count + 1);}return channel;}/*** 重连** @param channel*/public void reConnect(Channel channel) {log.info("{} 通道连接断开,正在重连...", channel);synchronized (channel) {if (channelList.indexOf(channel) == -1) {return;}channelList.set(channelList.indexOf(channel), bootstrap.connect().channel());}}

ChannelInitalize

public class UnvChannelInitialize extends ChannelInitializer {@Resourceprivate UnvHeartBeatHandler heartBeatHandler;@Resourceprivate SendMsgHandler sendMsgHandler;@Overrideprotected void initChannel(Channel channel) throws Exception {channel.pipeline().addLast(new StringEncoder(Charset.forName("UTF-8"))).addLast(new StringDecoder(Charset.forName("UTF-8")))//自定义解码器支持宇视协议.addLast("unvMsgEnCode", new UnvEncode()).addLast(heartBeatHandler).addLast(sendMsgHandler);}
}

Handler

```java
public class SendMsgHandler extends SimpleChannelInboundHandler {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {for (String pathKey : BlockingQueueConfig.queueMap.keySet()) {SendTask task = new SendTask(pathKey);InitServer.threadPoolExecutor.execute(task);}ctx.fireChannelActive();}@Overridepublic void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {// 不作处理ReferenceCountUtil.release(msg);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {log.error("通道连接异常:", cause);ctx.close();}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {log.error("{}:通道连接断开", ctx.channel());ctx.close();}@Overridepublic void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {super.channelWritabilityChanged(ctx);}
}

### 发送数据```java
public class SendTask implements Runnable {private FilePathKeyProperties pathKeyProperties;private NettyTcpClient tcpClient;private Map<String, FilePathKeyProperties.PathProperties> branchPath;private String pathKey;/*** 宇视物联协议报文命令字*/private int command;public SendTask(String pathKey) {this.pathKeyProperties = ApplicationContextUtil.getBean(FilePathKeyProperties.class);this.tcpClient = ApplicationContextUtil.getBean(NettyTcpClient.class);this.branchPath = pathKeyProperties.getBranchPath();this.pathKey = pathKey;}@Overridepublic void run() {File file = null;while (true) {try {BlockingQueue<File> fileQueue = BlockingQueueConfig.queueMap.get(pathKey);log.info("{}:阻塞队列大小:{}", pathKey, fileQueue.size());file = fileQueue.take();if (null == file || !file.exists()) {continue;}String fileName = file.getName();// 读取文件内容byte[] fileData = FileUtils.readFileToByteArray(file);// 实际文件中数据长度String length = fileName.split("\\.")[0].split(ConstantUtil.SPLITTER)[4];int fileDataLength = Integer.parseInt(length);// 文件内容为空或者长度为0,不发送if (null == fileData || fileData.length == 0) {file.delete();continue;}if (fileData.length < fileDataLength) {log.info("读取到的数据大小:{}, 实际文件数据大小:{}", fileData.length, fileDataLength);renameFile(file);continue;}command = branchPath.get(pathKey).getCommand();byte[] sendData = UnvProtocolUtil.buildUnvMsg(fileData, command);log.info("{}:数据大小:{}", branchPath.get(pathKey), sendData.length);Channel channel = tcpClient.getNextChannel();if (channel.isWritable()) {File tempFile = file;channel.writeAndFlush(sendData).addListener(future -> {if (!future.isSuccess()) {log.error("{}文件数据发送数据失败,重命名为.fail文件");renameFile(tempFile);} else {// 数据发送成功后,删除文件log.debug("{}:数据发送成功", tempFile.getName());if (!tempFile.delete()) {log.error("{}:文件删除失败", tempFile.getPath());}}});} else {renameFile(file);}} catch (InterruptedException e) {log.error("数据出队异常:", e);} catch (IOException e) {log.error("读取文件内容异常:", e);renameFile(file);} catch (Exception e) {log.error("数据发送异常:", e);renameFile(file);}}}/*** 将.temp文件重命名为.fail文件** @param file 文件*/private void renameFile(File file) {Runnable renameFailFile = () -> {try {int times = 0;String filePath = file.getPath();filePath = filePath.replace(ConstantUtil.TEMP_SUFFIX, ConstantUtil.FAIL_SUFFIX);File failFile = new File(filePath);while (true) {// 文件重命名成功,跳出循环if (file.renameTo(failFile)) {break;}// 文件重命名失败,延时0.5秒重试,失败三次后,文件删除if (times >= 2) {log.info("{}文件三次重命名失败,删除", filePath);file.delete();break;}times++;TimeUnit.MILLISECONDS.sleep(500);}log.info("将.temp文件重命名为{}", filePath);} catch (Exception e) {log.error("{}文件重命名为fail文件异常:", e);}};InitServer.failFileThreadPoolExecutor.execute(renameFailFile);}
}