wordpress块引用,郑州关键词优化费用,seo建站系统,沙井做网站公司背景 因为安装了正向隔离网闸#xff0c;导致数据传输的时候仅支持TCP协议和UDP协议#xff0c;因此需要开发UDP Client和Server服务来将数据透传#xff0c;当前环境是获取的数据并将数据转发到kafka PS: TCP 协议也能解决该问题#xff0c;但是TCP可能会出现粘包或者是半… 背景 因为安装了正向隔离网闸导致数据传输的时候仅支持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 ChannelInitializerDatagramChannel() {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 KafkaTemplateString, 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 ChannelInitializerDatagramChannel() {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 SimpleChannelInboundHandlerMapString, ByteBuf {//处理服务端返回的数据Overrideprotected void channelRead0(ChannelHandlerContext ctx, MapString, 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数据调试工具进行调试
