利用netty实现websocket ；redis的订阅发布websocket相结合

2024-08-22 05:44
文章标签 实现 redis websocket 发布 订阅 netty 相结合

本文主要是介绍利用netty实现websocket ；redis的订阅发布websocket相结合，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！ 

由于Http协议是无状态的，每一次请求只能响应一次，下次请求需要重新连接。
如果客户端请求一个服务端资源，需要实时监服务端执行状态（比如导出大数据量时需要前端监控导出状态），这个时候不断请求连接浪费资源。可以通过WebSocket建立一个长连接，实现客户端与服务端双向交流。

websocket服务器

public class NioWebSocketServer {private final Logger logger=Logger.getLogger(this.getClass());private void init(){logger.info("正在启动websocket服务器");NioEventLoopGroup boss=new NioEventLoopGroup();NioEventLoopGroup work=new NioEventLoopGroup();try {ServerBootstrap bootstrap=new ServerBootstrap();bootstrap.group(boss,work);bootstrap.channel(NioServerSocketChannel.class);bootstrap.childHandler(new NioWebSocketChannelInitializer());Channel channel = bootstrap.bind(8083).sync().channel();logger.info("webSocket服务器启动成功："+channel);channel.closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();logger.info("运行出错："+e);}finally {boss.shutdownGracefully();work.shutdownGracefully();logger.info("websocket服务器已关闭");}}public static void main(String[] args) {new NioWebSocketServer().init();}
}

ChannelInitializer

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;public class NioWebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) {ch.pipeline().addLast("logging",new LoggingHandler("DEBUG"));//设置log监听器，并且日志级别为debug，方便观察运行流程ch.pipeline().addLast("http-codec",new HttpServerCodec());//设置解码器ch.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));//聚合器，使用websocket会用到ch.pipeline().addLast("http-chunked",new ChunkedWriteHandler());//用于大数据的分区传输ch.pipeline().addLast("handler",new NioWebSocketHandler());//自定义的业务handler}
}

Hander

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import org.apache.log4j.Logger;
import org.wisdom.netty.global.ChannelSupervise;import java.util.Date;import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;public class NioWebSocketHandler extends SimpleChannelInboundHandler<Object> {private final Logger logger=Logger.getLogger(this.getClass());private WebSocketServerHandshaker handshaker;/**重写channelRead0方法，处理接收到的消息* @param ctx* @param msg* @throws Exception*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {logger.debug("收到消息："+msg);if (msg instanceof FullHttpRequest){//以http请求形式接入，但是走的是websockethandleHttpRequest(ctx, (FullHttpRequest) msg);}else if (msg instanceof  WebSocketFrame){//处理websocket客户端的消息handlerWebSocketFrame(ctx, (WebSocketFrame) msg);}}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//添加连接logger.debug("客户端加入连接："+ctx.channel());ChannelSupervise.addChannel(ctx.channel());}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {//断开连接logger.debug("客户端断开连接："+ctx.channel());ChannelSupervise.removeChannel(ctx.channel());}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame){// 判断是否关闭链路的指令if (frame instanceof CloseWebSocketFrame) {handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());return;}// 判断是否ping消息if (frame instanceof PingWebSocketFrame) {ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));return;}// 本例程仅支持文本消息，不支持二进制消息if (!(frame instanceof TextWebSocketFrame)) {logger.debug("本例程仅支持文本消息，不支持二进制消息");throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));}// 返回应答消息String request = ((TextWebSocketFrame) frame).text();logger.debug("服务端收到：" + request);TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString()+ ctx.channel().id() + "：" + request);// 群发ChannelSupervise.send2All(tws);// 返回【谁发的发给谁】// ctx.channel().writeAndFlush(tws);}/*** 唯一的一次http请求，用于创建websocket* */private void handleHttpRequest(ChannelHandlerContext ctx,FullHttpRequest req) {//要求Upgrade为websocket，过滤掉get/Postif (!req.decoderResult().isSuccess()|| (!"websocket".equals(req.headers().get("Upgrade")))) {//若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));return;}WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:8083/websocket", null, false);handshaker = wsFactory.newHandshaker(req);if (handshaker == null) {WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());} else {handshaker.handshake(ctx.channel(), req);}}/*** 拒绝不合法的请求，并返回错误信息* */private static void sendHttpResponse(ChannelHandlerContext ctx,FullHttpRequest req, DefaultFullHttpResponse res) {// 返回应答给客户端if (res.status().code() != 200) {ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(),CharsetUtil.UTF_8);res.content().writeBytes(buf);buf.release();}ChannelFuture f = ctx.channel().writeAndFlush(res);// 如果是非Keep-Alive，关闭连接if (!isKeepAlive(req) || res.status().code() != 200) {f.addListener(ChannelFutureListener.CLOSE);}}
}

存储信息

import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;public class ChannelSupervise {private   static ChannelGroup GlobalGroup=new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);private  static ConcurrentMap<String, ChannelId> ChannelMap=new ConcurrentHashMap();public  static void addChannel(Channel channel){/**channel(通道是socket的连接信息;) = [id: 0x2a5ba781, L:/127.0.0.1:8083 - R:/127.0.0.1:60663]==ChannelMap=={28176dd1=28176dd1, 2a5ba781=2a5ba781}*/GlobalGroup.add(channel);ChannelMap.put(channel.id().asShortText(),channel.id());System.out.println("==channel=="+channel);System.out.println("==ChannelMap=="+ChannelMap.toString());}public static void removeChannel(Channel channel){GlobalGroup.remove(channel);ChannelMap.remove(channel.id().asShortText());System.out.println("==removeChannel=="+channel);System.out.println("==removeChannelChannelMap=="+ChannelMap.toString());}public static  Channel findChannel(String id){return GlobalGroup.find(ChannelMap.get(id));}/*** 根据channel id 进行 群发通知* @param tws*/public static void send2All(TextWebSocketFrame tws){GlobalGroup.writeAndFlush(tws);}
}

html

<!-- index.html --><!DOCTYPE html>
<html>
<head><meta charset="UTF-8"><title>WebSocket Test</title>
</head>
<body><h1>WebSocket Test</h1><div><input type="text" id="message" placeholder="Message"><button onclick="send()">Send</button></div><div id="output"></div><script>var socket = new WebSocket("ws://localhost:8083/");socket.onopen = function(event) {console.log("WebSocket opened: " + event);};socket.onmessage = function(event) {console.log("WebSocket message received: " + event.data);var output = document.getElementById("output");output.innerHTML += "<p>" + event.data + "</p>";};socket.onclose = function(event) {console.log("WebSocket closed: " + event);};function send() {var message = document.getElementById("message").value;socket.send(message);}</script>
</body>
</html>

redis 订阅

   public void sendAlarmFaultMessage(String message) {String newMessge= null;try {newMessge = new String(message.getBytes(RedisKeyConstant.UTF8), RedisKeyConstant.UTF8);} catch (UnsupportedEncodingException e) {e.printStackTrace();}//redisTemplate.convertAndSend(RedisKeyConstant.REDIS_CHANNEL, newMessge);RedissonClient redissonClient = SpringUtil.getBean(RedissonClient.class);RTopic topic = redissonClient.getTopic(RedisKeyConstant.REDIS_CHANNEL_FAULT);topic.publish(newMessge);redisTemplate.opsForList().rightPush(RedisKeyConstant.REDIS_MESSAGE_FAULT, newMessge);}

redis发布

@Beanpublic RTopic rFaultTopic(RedissonClient redissonClient) {RTopic rTopic = redissonClient.getTopic(RedisKeyConstant.REDIS_CHANNEL_FAULT);try{if(rTopic != null){rTopic.addListener(String.class, (channel, message) -> {if (channel.toString().contains(RedisKeyConstant.REDIS_CHANNEL_FAULT)) {RedisUtil.lpop(RedisKeyConstant.REDIS_MESSAGE_FAULT);log.info("Channel is already fault "+message);AlarmWebsocketService alarmWebsocketService = SpringUtil.getBean(AlarmWebsocketService.class);alarmWebsocketService.sendAllMessage(message);}});}}catch (Exception e){log.info("Error sending alarm "+ ExceptionUtils.getStackTrace(e));}return rTopic;}

这篇关于利用netty实现websocket ；redis的订阅发布websocket相结合的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1095403。

相关文章

Spring Boot中WebSocket常用使用方法详解

Spring Boot中WebSocket常用使用方法详解

《SpringBoot中WebSocket常用使用方法详解》本文从WebSocket的基础概念出发,详细介绍了SpringBoot集成WebSocket的步骤,并重点讲解了常用的使用方法,包括简单消... 目录一、WebSocket基础概念1.1 什么是WebSocket1.2 WebSocket与HTTP
阅读更多...
Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

《Knife4j+Axios+Redis前后端分离架构下的API管理与会话方案(最新推荐)》本文主要介绍了Swagger与Knife4j的配置要点、前后端对接方法以及分布式Session实现原理,... 目录一、Swagger 与 Knife4j 的深度理解及配置要点Knife4j 配置关键要点1.Spri
阅读更多...
Python使用python-can实现合并BLF文件

Python使用python-can实现合并BLF文件

《Python使用python-can实现合并BLF文件》python-can库是Python生态中专注于CAN总线通信与数据处理的强大工具,本文将使用python-can为BLF文件合并提供高效灵活... 目录一、python-can 库：CAN 数据处理的利器二、BLF 文件合并核心代码解析1. 基础合
阅读更多...
Python使用OpenCV实现获取视频时长的小工具

Python使用OpenCV实现获取视频时长的小工具

《Python使用OpenCV实现获取视频时长的小工具》在处理视频数据时,获取视频的时长是一项常见且基础的需求,本文将详细介绍如何使用Python和OpenCV获取视频时长,并对每一行代码进行深入解析... 目录一、代码实现二、代码解析1. 导入 OpenCV 库2. 定义获取视频时长的函数3. 打开视频文
阅读更多...
golang版本升级如何实现

golang版本升级如何实现

《golang版本升级如何实现》：本文主要介绍golang版本升级如何实现问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录golanwww.chinasem.cng版本升级linux上golang版本升级删除golang旧版本安装golang最新版本总结gola
阅读更多...
SpringBoot中SM2公钥加密、私钥解密的实现示例详解

SpringBoot中SM2公钥加密、私钥解密的实现示例详解

《SpringBoot中SM2公钥加密、私钥解密的实现示例详解》本文介绍了如何在SpringBoot项目中实现SM2公钥加密和私钥解密的功能,通过使用Hutool库和BouncyCastle依赖,简化... 目录一、前言1、加密信息(示例)2、加密结果(示例)二、实现代码1、yml文件配置2、创建SM2工具
阅读更多...
Mysql实现范围分区表(新增、删除、重组、查看)

Mysql实现范围分区表(新增、删除、重组、查看)

《Mysql实现范围分区表(新增、删除、重组、查看)》MySQL分区表的四种类型（范围、哈希、列表、键值）,主要介绍了范围分区的创建、查询、添加、删除及重组织操作,具有一定的参考价值,感兴趣的可以了解... 目录一、mysql分区表分类二、范围分区（Range Partitioning1、新建分区表：2、分
阅读更多...
MySQL 定时新增分区的实现示例

MySQL 定时新增分区的实现示例

《MySQL定时新增分区的实现示例》本文主要介绍了通过存储过程和定时任务实现MySQL分区的自动创建,解决大数据量下手动维护的繁琐问题,具有一定的参考价值,感兴趣的可以了解一下... mysql创建好分区之后，有时候会需要自动创建分区。比如，一些表数据量非常大，有些数据是热点数据，按照日期分区MululbU
阅读更多...
MySQL中查找重复值的实现

MySQL中查找重复值的实现

《MySQL中查找重复值的实现》查找重复值是一项常见需求,比如在数据清理、数据分析、数据质量检查等场景下,我们常常需要找出表中某列或多列的重复值,具有一定的参考价值,感兴趣的可以了解一下... 目录技术背景实现步骤方法一：使用GROUP BY和HAVING子句方法二：仅返回重复值方法三：返回完整记录方法四：
阅读更多...
IDEA中新建/切换Git分支的实现步骤

IDEA中新建/切换Git分支的实现步骤

《IDEA中新建/切换Git分支的实现步骤》本文主要介绍了IDEA中新建/切换Git分支的实现步骤,通过菜单创建新分支并选择是否切换,创建后在Git详情或右键Checkout中切换分支,感兴趣的可以了... 前提：项目已被Git托管1、点击上方栏Git->NewBrancjsh...2、输入新的分支的
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2