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

网站内页关键词密度crm系统

news 2025/8/15 16:57:52

网站内页关键词密度,crm系统,做视频网站教程,合肥做网站背景因为安装了正向隔离网闸，导致数据传输的时候仅支持TCP协议和UDP协议，因此需要开发UDP Client和Server服务来将数据透传，当前环境是获取的数据并将数据转发到kafka PS: TCP 协议也能解决该问题，但是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 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();}
}