Spring Boot+Netty

ops/2024/12/29 2:08:41/

因工作中需要给第三方屏幕厂家下发广告，音频，图片等内容，对方提供TCP接口于是我使用Netty长链接进行数据传输

1.添加依赖

 <!--  netty依赖--><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId></dependency>

2.创建Netty服务

@Slf4j
@Component
public class NettyServer {public void start(InetSocketAddress address) {//配置服务端的NIO线程组EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {// 绑定线程池,编码解码//服务端接受连接的队列长度，如果队列已满，客户端连接将被拒绝ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup)// 指定Channel.channel(NioServerSocketChannel.class)//使用指定的端口设置套接字地址.localAddress(address)//使用自定义处理类.childHandler(new NettyServerChannelInitializer())//服务端可连接队列数,对应TCP/IP协议listen函数中backlog参数.option(ChannelOption.SO_BACKLOG, 128)//保持长连接，2小时无数据激活心跳机制.childOption(ChannelOption.SO_KEEPALIVE, true)//将小的数据包包装成更大的帧进行传送，提高网络的负载.childOption(ChannelOption.TCP_NODELAY, true);// 绑定端口，开始接收进来的连接ChannelFuture future = bootstrap.bind(address).sync();if (future.isSuccess()) {log.info("netty服务器开始监听端口：{}",address.getPort());}//关闭channel和块，直到它被关闭future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

3.创建Socket配置类（也可以直接在步骤2中写死）

1.在配置文件中

socket:# 监听端口 8090port: 8090#ip地址host: 0.0.0.0
#  host: 192.168.31.2@Setter
@Getter
@ToString
@Component
@Configuration
@PropertySource("classpath:application.yml")
@ConfigurationProperties(prefix = "socket")
public class SocketProperties {private Integer port;private String host;}

4.在springboot 启动类中启用Netty服务

@SpringBootApplication
public class Application implements CommandLineRunner {public static void main(String[] args) {SpringApplication application = new SpringApplication(Application.class);application.setApplicationStartup(new BufferingApplicationStartup(2048));application.run(args);}@Resourceprivate NettyServer nettyServer;@Resourceprivate SocketProperties socketProperties;@Overridepublic void run(String... args) {InetSocketAddress address = new InetSocketAddress(socketProperties.getHost(),socketProperties.getPort());nettyServer.start(address);}}

5.创建字符解析器，解析收到的消息

/*** 功能描述: 服务端初始化，客户端与服务器端连接一旦创建，这个类中方法就会被回调，设置出站编码器和入站解码器**/
public class NettyServerChannelInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();//接收消息格式,使用自定义解析数据格式
//        pipeline.addLast("decoder",new MyDecoder());//发送消息格式，使用自定义解析数据格式
//        pipeline.addLast("encoder",new MyEncoder());pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));//针对客户端，如果在1分钟时没有想服务端发送写心跳(ALL)，则主动断开//如果是读空闲或者写空闲，不处理,这里根据自己业务考虑使用pipeline.addLast(new IdleStateHandler(0,0,90, TimeUnit.SECONDS));//自定义的空闲检测pipeline.addLast(new NettyServerHandler());}
}

6.创建Handler 类处理消息

/*** 功能描述: netty服务端处理类*/@Slf4j
@Component
public class NettyServerHandler extends ChannelInboundHandlerAdapter {/*** 功能描述: 有客户端连接服务器会触发此函数** @param ctx 通道* @return void*/@Overridepublic void channelActive(ChannelHandlerContext ctx) {InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();String clientIp = insocket.getAddress().getHostAddress();int clientPort = insocket.getPort();//获取连接通道唯一标识ChannelId channelId = ctx.channel().id();//如果map中不包含此连接，就保存连接if (ChannelMap.getChannelMap().containsKey(channelId)) {log.info("客户端:{},是连接状态，连接通道数量:{} ", channelId, ChannelMap.getChannelMap().size());} else {//保存连接ChannelMap.addChannel(channelId, ctx.channel());log.info("客户端:{},连接netty服务器[IP:{}-->PORT:{}]", channelId, clientIp, clientPort);log.info("连接通道数量: {}", ChannelMap.getChannelMap().size());}}/*** 功能描述: 有客户端终止连接服务器会触发此函数* @param ctx 通道处理程序上下文* @return void*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) {InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();String clientIp = inSocket.getAddress().getHostAddress();ChannelId channelId = ctx.channel().id();//包含此客户端才去删除if (ChannelMap.getChannelMap().containsKey(channelId)) {//删除连接ChannelMap.getChannelMap().remove(channelId);log.info("客户端:{},断开netty服务器[IP:{}-->PORT:{}]", channelId, clientIp, inSocket.getPort());log.info("连接通道数量: " + ChannelMap.getChannelMap().size());}}@Transactional@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;ByteBuf rebuf = Unpooled.buffer();RedisUtils.setChannelId(ctx.channel().id().toString(), ctx.channel().id());// 读取帧头标识byte frameHeader = buf.readByte();if (frameHeader != 0x7E) {byte[] data = ByteBufUtil.getBytes(buf);String hex = bytesToHex(data);buf.release();String content = ((ByteBuf) msg).toString(Charset.defaultCharset());}  // 读取消息帧类型else {byte messageType = buf.readByte();// 读取帧尾标识if (buf.isReadable()) {// 读取校验值byte checksum = buf.readByte();byte frameTail = buf.readByte();}}buf.release();}/*** 功能描述: 服务端给客户端发送消息** @param channelId 连接通道唯一id* @param msg       需要发送的消息内容* @return void*/public void channelWrite(ChannelId channelId, Object msg) throws Exception {Channel channel = ChannelMap.getChannelMap().get(channelId);if (channel == null) {log.info("通道:{},不存在", channelId);return;}if (msg == null || msg == "") {log.info("服务端响应空的消息");return;}//将客户端的信息直接返回写入ctxchannel.write(msg);//刷新缓存区channel.flush();}@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {String socketString = ctx.channel().remoteAddress().toString();if (evt instanceof IdleStateEvent) {IdleStateEvent event = (IdleStateEvent) evt;if (event.state() == IdleState.READER_IDLE) {log.info("Client:{},READER_IDLE 读超时", socketString);Channel channel = ctx.channel();ChannelId id = channel.id();// 超时未收到心跳包，更新设备状态为离线// todo 更新设备状态ctx.disconnect();ChannelMap.removeChannelByName(id);} else if (event.state() == IdleState.WRITER_IDLE) {log.info("Client:{}, WRITER_IDLE 写超时", socketString);ctx.disconnect();Channel channel = ctx.channel();ChannelId id = channel.id();ChannelMap.removeChannelByName(id);} else if (event.state() == IdleState.ALL_IDLE) {log.info("Client:{},ALL_IDLE 总超时", socketString);Channel channel = ctx.channel();ChannelId id = channel.id();// 超时未收到心跳包，更新设备状态为离线// todo 更新设备状态ctx.disconnect();ChannelMap.removeChannelByName(id);}}}/*** 功能描述: 发生异常会触发此函数** @param ctx   通道处理程序上下文* @param cause 异常* @return void*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();log.info("{}:发生了错误,此连接被关闭。此时连通数量:{}", ctx.channel().id(), ChannelMap.getChannelMap().size());}}

ChannelMap类

/*** 功能描述: 管理通道Map类**/
public class ChannelMap {/*** 管理一个全局map，保存连接进服务端的通道数量*/private static final ConcurrentHashMap<ChannelId, Channel> CHANNEL_MAP = new ConcurrentHashMap<>(128);public static ConcurrentHashMap<ChannelId, Channel> getChannelMap() {return CHANNEL_MAP;}/***  获取指定name的channel*/public static Channel getChannelByName(ChannelId channelId){if(CollectionUtils.isEmpty(CHANNEL_MAP)){return null;}return CHANNEL_MAP.get(channelId);}/***  将通道中的消息推送到每一个客户端*/public static boolean pushNewsToAllClient(String obj){if(CollectionUtils.isEmpty(CHANNEL_MAP)){return false;}for(ChannelId channelId: CHANNEL_MAP.keySet()) {Channel channel = CHANNEL_MAP.get(channelId);channel.writeAndFlush(new TextWebSocketFrame(obj));}return true;}/***  将channel和对应的name添加到ConcurrentHashMap*/public static void addChannel(ChannelId channelId,Channel channel){CHANNEL_MAP.put(channelId,channel);}/***  移除掉name对应的channel*/public static boolean removeChannelByName(ChannelId channelId){if(CHANNEL_MAP.containsKey(channelId)){CHANNEL_MAP.remove(channelId);return true;}return false;}}

http://www.ppmy.cn/ops/139586.html

相关文章

EasyAnimateV5 视频生成大模型原理详解与模型使用

EasyAnimateV5 视频生成大模型原理详解与模型使用

在数字内容创作中&#xff0c;视频扮演的角色日益重要。然而&#xff0c;创作高质量视频通常耗时且昂贵。EasyAnimate 系列旨在利用人工智能技术简化这一过程。EasyAnimateV5 建立在其前代版本的基础之上&#xff0c;不仅在质量上有所提升&#xff0c;还在多模态数据处理和跨语…
阅读更多...
【附源码】基于环信鸿蒙IM SDK实现一个聊天Demo

【附源码】基于环信鸿蒙IM SDK实现一个聊天Demo

项目背景 本项目基于环信IM 鸿蒙SDK 打造的鸿蒙IM Demo&#xff0c;完全适配HarmonyOS NEXT系统&#xff0c;实现了发送消息&#xff0c;添加好友等基础功能。代码开源&#xff0c;功能简洁&#xff0c;如果您有类似开发需求可以参考。 源码地址&#xff1a;https://github.c…
阅读更多...
Word2vec、词向量是什么？ |Gensim中word2vec模型的参数定义

Word2vec、词向量是什么？ |Gensim中word2vec模型的参数定义

前言&#xff1a; 最近在忙毕设&#xff0c;要学习一些AI的技术。很多资料看来看去&#xff0c;感觉只是在大脑皮层表面略过了一下&#xff0c;遂还是决定采用老方法&#xff0c;写博客&#xff01;&#xff01;&#xff01;对了&#xff0c;我也只是一个萌新&#xff0c;博客的…
阅读更多...
网络原理之 UDP 协议

网络原理之 UDP 协议

目录 1. UDP 协议报文格式 2. UDP 的特点 (1) 无连接 (2) 不可靠 (3) 面向数据报 (4) 全双工 3. 基于 UDP 的应用层协议 前文是&#xff1a;UDP 的使用 首先了解一下基础知识&#xff1a; 1. UDP 协议报文格式 传输层最重要的协议有两个&#xff0c;一个是 TCP&#x…
阅读更多...
vscode通过ssh连接虚拟机进行开发

vscode通过ssh连接虚拟机进行开发

虚拟机自带的vscode很卡而且画质感觉不行&#xff0c;所以用这种方法解决 1.VSCODE安装扩展Tabnine(AI代码补全)&#xff0c;Remote Development 2.虚拟机终端ifconfig查看本机ip 192.168.43.197 开启ubuntu的SSH服务 sudo apt-get install openssh-server 配置vscode的ssh …
阅读更多...
登Nature子刊！华中师范大学提出DigFrag，用AI精准分割分子片段，并生成44个药物/农药分子

登Nature子刊！华中师范大学提出DigFrag，用AI精准分割分子片段，并生成44个药物/农药分子

过去几十年&#xff0c;基于片段的药物发现 (FBDD) 通过识别与靶标蛋白有微弱相互作用的小分子片段&#xff0c;并优化这些片段的结构信息&#xff0c;可以开发出活性更高的先导化合物&#xff0c;在新药研发中发挥了重要作用。 尽管 FBDD 在药物发现和开发领域扮演着关键角色…
阅读更多...
springai结合ollama

springai结合ollama

目录 ollama 介绍 使用 下载&#xff1a; 安装&#xff1a; 点击这个玩意next就行了。 运行 spring ai使用ollama调用本地部署的大模型 加依赖 配置yml 写代码 ollama 介绍 官网&#xff1a;Ollama Ollama是一个用于部署和运行各种开源大模型的工具&#xff1b; …
阅读更多...
热点数据失效 详解

热点数据失效 详解

热点数据失效详解 热点数据失效是指在分布式系统中&#xff0c;缓存中某些被频繁访问的热点数据在某个时间点同时失效&#xff0c;导致大量请求直接涌向数据库或后端服务&#xff0c;造成系统瞬间高负载甚至宕机。热点数据失效问题常见于高并发的业务场景&#xff08;如秒杀、…
阅读更多...
最新文章

Copyright @ 2022~2023