利用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生成随机唯一id的几种实现方法

python生成随机唯一id的几种实现方法

《python生成随机唯一id的几种实现方法》在Python中生成随机唯一ID有多种方法,根据不同的需求场景可以选择最适合的方案,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习... 目录方法 1：使用 UUID 模块（推荐）方法 2：使用 Secrets 模块（安全敏感场景）方法
阅读更多...
Redis中Stream详解及应用小结

Redis中Stream详解及应用小结

《Redis中Stream详解及应用小结》RedisStreams是Redis5.0引入的新功能,提供了一种类似于传统消息队列的机制,但具有更高的灵活性和可扩展性,本文给大家介绍Redis中Strea... 目录1. Redis Stream 概述2. Redis Stream 的基本操作2.1. XADD
阅读更多...
Spring StateMachine实现状态机使用示例详解

Spring StateMachine实现状态机使用示例详解

《SpringStateMachine实现状态机使用示例详解》本文介绍SpringStateMachine实现状态机的步骤,包括依赖导入、枚举定义、状态转移规则配置、上下文管理及服务调用示例,重点解... 目录什么是状态机使用示例什么是状态机状态机是计算机科学中的​​核心建模工具​​，用于描述对象在其生命
阅读更多...
Spring Boot 结合 WxJava 实现文章上传微信公众号草稿箱与群发

Spring Boot 结合 WxJava 实现文章上传微信公众号草稿箱与群发

《SpringBoot结合WxJava实现文章上传微信公众号草稿箱与群发》本文将详细介绍如何使用SpringBoot框架结合WxJava开发工具包,实现文章上传到微信公众号草稿箱以及群发功能,... 目录一、项目环境准备1.1 开发环境1.2 微信公众号准备二、Spring Boot 项目搭建2.1 创建
阅读更多...
IntelliJ IDEA2025创建SpringBoot项目的实现步骤

IntelliJ IDEA2025创建SpringBoot项目的实现步骤

《IntelliJIDEA2025创建SpringBoot项目的实现步骤》本文主要介绍了IntelliJIDEA2025创建SpringBoot项目的实现步骤,文中通过示例代码介绍的非常详细,对大家... 目录一、创建 Spring Boot 项目1. 新建项目2. 基础配置3. 选择依赖4. 生成项目5.
阅读更多...
Linux下删除乱码文件和目录的实现方式

Linux下删除乱码文件和目录的实现方式

《Linux下删除乱码文件和目录的实现方式》：本文主要介绍Linux下删除乱码文件和目录的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux下删除乱码文件和目录方法1方法2总结Linux下删除乱码文件和目录方法1使用ls -i命令找到文件或目录
阅读更多...
SpringBoot+EasyExcel实现自定义复杂样式导入导出

SpringBoot+EasyExcel实现自定义复杂样式导入导出

《SpringBoot+EasyExcel实现自定义复杂样式导入导出》这篇文章主要为大家详细介绍了SpringBoot如何结果EasyExcel实现自定义复杂样式导入导出功能,文中的示例代码讲解详细,... 目录安装处理自定义导出复杂场景1、列不固定，动态列2、动态下拉3、自定义锁定行/列，添加密码4、合并
阅读更多...
mybatis执行insert返回id实现详解

mybatis执行insert返回id实现详解

《mybatis执行insert返回id实现详解》MyBatis插入操作默认返回受影响行数,需通过useGeneratedKeys+keyProperty或selectKey获取主键ID,确保主键为自... 目录 两种方式获取自增 ID：1. ​​useGeneratedKeys+keyProperty（推
阅读更多...
Spring Boot集成Druid实现数据源管理与监控的详细步骤

Spring Boot集成Druid实现数据源管理与监控的详细步骤

《SpringBoot集成Druid实现数据源管理与监控的详细步骤》本文介绍如何在SpringBoot项目中集成Druid数据库连接池,包括环境搭建、Maven依赖配置、SpringBoot配置文件... 目录1. 引言1.1 环境准备1.2 Druid介绍2. 配置Druid连接池3. 查看Druid监控
阅读更多...
Linux在线解压jar包的实现方式

Linux在线解压jar包的实现方式

《Linux在线解压jar包的实现方式》：本文主要介绍Linux在线解压jar包的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux在线解压jar包解压 jar包的步骤总结Linux在线解压jar包在 Centos 中解压 jar 包可以使用 u
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2