当前位置： 首页 > news >正文

基于Netty的UDPServer端和Client端解决正向隔离网闸数据透传问题

news 2025/6/14 7:00:57

背景

因为安装了正向隔离网闸，导致数据传输的时候仅支持TCP协议和UDP协议，因此需要开发UDP Client和Server服务来将数据透传，当前环境是获取的数据并将数据转发到kafka

PS: TCP 协议也能解决该问题，但是TCP可能会出现粘包或者是半包问题

 1.UDP Server端

 server端启动类

package com.huanyu.forward.udp.server;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramChannel;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;@Slf4j
@Service("udpServer")
@ConditionalOnExpression("#{'${spring.udp-server.port:}'.length()>0}")
public class UdpNettyServer {@Value("${spring.udp-server.port:33333}")private Integer port = 33333;public static void main(String[] args) throws Exception {new UdpNettyServer().udpServer(33333);}@PostConstruct()public void initUdpServer() {try {log.info("start udp server......");udpServer(port);} catch (Exception e) {log.error("tcp server start failed");}}public void udpServer(int port) throws Exception {EventLoopGroup group = new NioEventLoopGroup();Bootstrap b = new Bootstrap();b.group(group).channel(NioDatagramChannel.class).option(ChannelOption.SO_BROADCAST, true).localAddress(new InetSocketAddress(port)).handler(new ChannelInitializer<DatagramChannel>() {@Overrideprotected void initChannel(DatagramChannel ch) {ChannelPipeline pipeline = ch.pipeline();
//                            pipeline.addLast(new LoggingHandler(LogLevel.DEBUG));pipeline.addLast(new UdpDataHandler());}});//绑定监听端口，调用sync同步阻塞方法等待绑定操作完ChannelFuture future = b.bind().sync();if (future.isSuccess()) {log.info("udp server is listening on  :{}", port);} else {log.error("udp server is failed ", future.cause());//关闭线程组group.shutdownGracefully();}//成功绑定到端口之后,给channel增加一个 管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程。
//            future.channel().closeFuture().sync();}
}

 Server数据处理类

package com.huanyu.forward.udp.server;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.DatagramPacket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;import java.nio.charset.StandardCharsets;@Slf4j
@Service
public class UdpDataHandler extends ChannelInboundHandlerAdapter {public static final String BIZ_FLAG = "bizFlag";public static final String FLAG_PRE = "@@{";public static final String FLAG_SUF = "}##";public static final byte[] FLAG_PREFIX = FLAG_PRE.getBytes(StandardCharsets.UTF_8);public static final byte[] FLAG_SUFFIX = FLAG_SUF.getBytes(StandardCharsets.UTF_8);//    @Resource
//    private KafkaTemplate<String, Object> template;//接受client发送的消息@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {try {DatagramPacket p = (DatagramPacket) msg;ByteBuf in = p.content();byte[] flagBytes = getBytes(in);if (flagBytes == null) {return;}in.readBytes(flagBytes); // 读取标志位// 保留标志位的对象结构-以@@{开头以}##结尾，形如@@{"k":"v"}##{"k":"v"}$,@@和##之间的数据为补充的对象参数JSON,$为换行符号String topicData = new String(flagBytes, FLAG_PRE.length() - 1, flagBytes.length - FLAG_PREFIX.length - FLAG_SUFFIX.length + 2, StandardCharsets.UTF_8);byte[] msgByte = new byte[in.readableBytes()];in.readBytes(msgByte);
//        template.send("haha.haha.ha", gbk.getBytes());log.info("bizFag:{},data: {}", topicData, new String(msgByte));} catch (Exception e) {log.error("udp handler 异常： ", e);}}private byte[] getBytes(ByteBuf in) {if (in.readableBytes() < FLAG_PREFIX.length + FLAG_SUFFIX.length) {log.warn("数据长度不够");text(in);return null;}int prefixIndex = in.readerIndex();if (!startsWith(in)) {text(in);// 忽略非标志位开头的数据in.skipBytes(in.readableBytes());log.warn("数据不包含指定的前缀");return null;}int suffixIndex = indexOf(in);if (suffixIndex == -1) {log.warn("数据不包含指定的某字符");text(in);return null;}int flagLength = suffixIndex - prefixIndex + FLAG_SUFFIX.length;return new byte[flagLength];}//通知处理器最后的channelRead()是当前批处理中的最后一条消息时调用@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}//读操作时捕获到异常时调用@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {ctx.close();}//客户端去和服务端连接成功时触发@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("hello udp client [你好，客户端]".getBytes()));log.info("udp client 连接成功： {}", ctx.channel());}private boolean startsWith(ByteBuf buf) {for (int i = 0; i < FLAG_PREFIX.length; i++) {if (buf.getByte(buf.readerIndex() + i) != FLAG_PREFIX[i]) {return false;}}return true;}private int indexOf(ByteBuf buf) {int readerIndex = buf.readerIndex();int readableBytes = buf.readableBytes();for (int i = 0; i <= readableBytes - FLAG_SUFFIX.length; i++) {boolean match = true;for (int j = 0; j < FLAG_SUFFIX.length; j++) {if (buf.getByte(readerIndex + i + j) != FLAG_SUFFIX[j]) {match = false;break;}}if (match) {return readerIndex + i;}}return -1;}private void text(ByteBuf in) {byte[] msgByte = new byte[in.readableBytes()];in.readBytes(msgByte);log.warn("数据：{}", new String(msgByte, StandardCharsets.UTF_8));}
}

 2.UDP Client端

 Client端启动类

package com.aimsphm.forward.udp.client;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramChannel;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.util.CharsetUtil;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.stream.IntStream;@Getter
@Slf4j
public class UdpNettyClient {private final InetSocketAddress net;private Channel channel;public UdpNettyClient(String host, int port) {net = new InetSocketAddress(host, port);udpClient();}public void udpClient() {try {EventLoopGroup group = new NioEventLoopGroup();Bootstrap b = new Bootstrap();b.group(group).channel(NioDatagramChannel.class).handler(new ChannelInitializer<DatagramChannel>() {@Overrideprotected void initChannel(DatagramChannel ch) {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new UdpClientHandler());}});ChannelFuture future = b.bind(0).sync();future.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture arg0) throws Exception {if (future.isSuccess()) {log.info("连接UDP服务器成功:");} else {log.warn("连接UDP服务器失败:");System.out.println("连接服务器失败");group.shutdownGracefully(); //关闭线程组}}});this.channel = future.channel(); // 绑定一个随机端口进行发送} catch (InterruptedException e) {log.error("UDP服务端启动异常：", e);}}public static void main(String[] args) throws Exception {UdpNettyClient client = new UdpNettyClient("localhost", 33333);Channel ch = client.getChannel(); // 修改为你的服务器地址和端口// 发送数据包IntStream.range(0, 2).parallel().forEach(i -> {try {ch.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("@@{\"k\":\"v" + i + "\"}##{\"k\":\"v\"}", CharsetUtil.UTF_8), client.getNet())).sync();} catch (InterruptedException e) {throw new RuntimeException(e);}});}
}

 Client数据处理类

package com.huanyu.forward.udp.client;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;import java.util.Map;public class UdpClientHandler extends SimpleChannelInboundHandler<Map<String, ByteBuf>> {//处理服务端返回的数据@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Map<String, ByteBuf> data) throws Exception {ByteBuf msg = data.get("topic");byte[] msgByte = new byte[msg.readableBytes()];msg.readBytes(msgByte);System.out.println("接受到server响应数据: " + new String(msgByte));}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("hello server 你好".getBytes()));super.channelActive(ctx);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}
}

分别启动服务端和客户端，进行数据调试吧，也可以下载TCP/UDP数据调试工具，进行调试

 

http://www.xdnf.cn/news/1019035.html

相关文章：

  • 前端八股文-vue篇
  • 2025-06-13【视频处理】基于视频内容转场进行分割
  • 深度剖析：AI 社媒矩阵营销工具，如何高效获客？
  • 实验复现：应用 RIR 触发器的 TrojanRoom 后门攻击实现
  • Java虚拟机解剖：从字节码到机器指令的终极之旅（二）
  • 【第一章:人工智能基础】03.算法分析与设计-(4)贪心算法（Greedy Algorithm）
  • C++ 中文件 IO 操作详解
  • 软件开发 | 从 Azure DevOps迁移至GitHub企业版的最佳路径
  • HTTP全攻略：从入门到精通
  • @RequestHeader(“Authorization“) 解析：HTTP 请求头中的 Authorization 字段
  • JSON 编辑器：从语法到数据处理（二）
  • 在C#中乐观锁的实现
  • ios 26发布：设计革新与智能整合
  • 分析实例，学习了解浏览器事件循环机制
  • 基于ssm的教学质量评估系统
  • CIM和建筑风貌管控平台
  • [7-01-03].第03节：环境搭建 - 集群架构
  • Java企业技术趋势分析：AI应用的落地实践与未来展望
  • nuxt2报错Unexpected token ‘{‘
  • CSS flex-basis 属性详解：功能、用法与最佳实践
  • CSS Houdini 解锁前端动画的下一个时代！
  • 主流版本控制工具Git vs Perforce P4：架构模式、性能、大文件管理及分支管理对比详解
  • 在线教程丨刷新TTS模型SOTA，OpenAudio S1基于200万小时音频数据训练，深刻理解情感及语音细节
  • 引入 Kafka 消息队列解耦热点操作
  • list使用及模拟
  • HarmonyOS 应用模块化设计 - 面试核心知识点
  • WPF--Application.Current.Dispatcher.BeginInvoke
  • 在Jupyter Notebook中使用Conda虚拟环境
  • 使用 PyMuPDF 和 PySide6/PyQt6 编写的 PDF 查看器 （显示树状书签和缩略图列表，没有文字选择功能）
  • Monte Carlo衍生品定价（金融工程）