利用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。

相关文章

使用Python实现IP地址和端口状态检测与监控

使用Python实现IP地址和端口状态检测与监控

《使用Python实现IP地址和端口状态检测与监控》在网络运维和服务器管理中,IP地址和端口的可用性监控是保障业务连续性的基础需求,本文将带你用Python从零打造一个高可用IP监控系统,感兴趣的小伙... 目录概述：为什么需要IP监控系统使用步骤说明1. 环境准备2. 系统部署3. 核心功能配置系统效果展
阅读更多...
Python实现微信自动锁定工具

Python实现微信自动锁定工具

《Python实现微信自动锁定工具》在数字化办公时代,微信已成为职场沟通的重要工具,但临时离开时忘记锁屏可能导致敏感信息泄露,下面我们就来看看如何使用Python打造一个微信自动锁定工具吧... 目录引言：当微信隐私遇到自动化守护效果展示核心功能全景图技术亮点深度解析1. 无操作检测引擎2. 微信路径智能获
阅读更多...
redis中使用lua脚本的原理与基本使用详解

redis中使用lua脚本的原理与基本使用详解

《redis中使用lua脚本的原理与基本使用详解》在Redis中使用Lua脚本可以实现原子性操作、减少网络开销以及提高执行效率,下面小编就来和大家详细介绍一下在redis中使用lua脚本的原理... 目录Redis 执行 Lua 脚本的原理基本使用方法使用EVAL命令执行 Lua 脚本使用EVALSHA命令
阅读更多...
Python中pywin32 常用窗口操作的实现

Python中pywin32 常用窗口操作的实现

《Python中pywin32常用窗口操作的实现》本文主要介绍了Python中pywin32常用窗口操作的实现,pywin32主要的作用是供Python开发者快速调用WindowsAPI的一个... 目录获取窗口句柄获取最前端窗口句柄获取指定坐标处的窗口根据窗口的完整标题匹配获取句柄根据窗口的类别匹配获取句
阅读更多...
在 Spring Boot 中实现异常处理最佳实践

在 Spring Boot 中实现异常处理最佳实践

《在SpringBoot中实现异常处理最佳实践》本文介绍如何在SpringBoot中实现异常处理,涵盖核心概念、实现方法、与先前查询的集成、性能分析、常见问题和最佳实践,感兴趣的朋友一起看看吧... 目录一、Spring Boot 异常处理的背景与核心概念1.1 为什么需要异常处理？1.2 Spring B
阅读更多...
Python位移操作和位运算的实现示例

Python位移操作和位运算的实现示例

《Python位移操作和位运算的实现示例》本文主要介绍了Python位移操作和位运算的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 位移操作1.1 左移操作 (<<)1.2 右移操作 (>>)注意事项：2. 位运算2.1
阅读更多...
如何在 Spring Boot 中实现 FreeMarker 模板

如何在 Spring Boot 中实现 FreeMarker 模板

《如何在SpringBoot中实现FreeMarker模板》FreeMarker是一种功能强大、轻量级的模板引擎,用于在Java应用中生成动态文本输出（如HTML、XML、邮件内容等）,本文... 目录什么是 FreeMarker 模板？在 Spring Boot 中实现 FreeMarker 模板1. 环
阅读更多...
Qt实现网络数据解析的方法总结

Qt实现网络数据解析的方法总结

《Qt实现网络数据解析的方法总结》在Qt中解析网络数据通常涉及接收原始字节流,并将其转换为有意义的应用层数据,这篇文章为大家介绍了详细步骤和示例,感兴趣的小伙伴可以了解下... 目录1. 网络数据接收2. 缓冲区管理（处理粘包/拆包）3. 常见数据格式解析3.1 jsON解析3.2 XML解析3.3 自定义
阅读更多...
SpringMVC 通过ajax 前后端数据交互的实现方法

SpringMVC 通过ajax 前后端数据交互的实现方法

《SpringMVC通过ajax前后端数据交互的实现方法》：本文主要介绍SpringMVC通过ajax前后端数据交互的实现方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价... 在前端的开发过程中，经常在html页面通过AJAX进行前后端数据的交互，SpringMVC的controll
阅读更多...
Redis 热 key 和大 key 问题小结

Redis 热 key 和大 key 问题小结

《Redis热key和大key问题小结》：本文主要介绍Redis热key和大key问题小结,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录一、什么是 Redis 热 key？热 key（Hot Key）定义： 热 key 常见表现：热 key 的风险：二、
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2