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

相关文章

Go语言开发实现查询IP信息的MCP服务器

Go语言开发实现查询IP信息的MCP服务器

《Go语言开发实现查询IP信息的MCP服务器》随着MCP的快速普及和广泛应用,MCP服务器也层出不穷,本文将详细介绍如何在Go语言中使用go-mcp库来开发一个查询IP信息的MCP... 目录前言mcp-ip-geo 服务器目录结构说明查询 IP 信息功能实现工具实现工具管理查询单个 IP 信息工具的实现服
阅读更多...
SpringBoot基于配置实现短信服务策略的动态切换

SpringBoot基于配置实现短信服务策略的动态切换

《SpringBoot基于配置实现短信服务策略的动态切换》这篇文章主要为大家详细介绍了SpringBoot在接入多个短信服务商（如阿里云、腾讯云、华为云）后,如何根据配置或环境切换使用不同的服务商,需... 目录目标功能示例配置（application.yml）配置类绑定短信发送策略接口示例：阿里云 & 腾
阅读更多...
Redis Pipeline(管道) 详解

Redis Pipeline(管道) 详解

《RedisPipeline(管道)详解》Pipeline管道是Redis提供的一种批量执行命令的机制,通过将多个命令一次性发送到服务器并统一接收响应,减少网络往返次数（RTT）,显著提升执行效率... 目录Redis Pipeline 详解1. Pipeline 的核心概念2. 工作原理与性能提升3. 核
阅读更多...
python实现svg图片转换为png和gif

python实现svg图片转换为png和gif

《python实现svg图片转换为png和gif》这篇文章主要为大家详细介绍了python如何实现将svg图片格式转换为png和gif,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录python实现svg图片转换为png和gifpython实现图片格式之间的相互转换延展：基于Py
阅读更多...
Python利用ElementTree实现快速解析XML文件

Python利用ElementTree实现快速解析XML文件

《Python利用ElementTree实现快速解析XML文件》ElementTree是Python标准库的一部分,而且是Python标准库中用于解析和操作XML数据的模块,下面小编就来和大家详细讲讲... 目录一、XML文件解析到底有多重要二、ElementTree快速入门1. 加载XML的两种方式2.
阅读更多...
Java的栈与队列实现代码解析

Java的栈与队列实现代码解析

《Java的栈与队列实现代码解析》栈是常见的线性数据结构,栈的特点是以先进后出的形式,后进先出,先进后出,分为栈底和栈顶,栈应用于内存的分配,表达式求值,存储临时的数据和方法的调用等,本文给大家介绍J... 目录栈的概念（Stack）栈的实现代码队列（Queue）模拟实现队列(双链表实现)循环队列（循环数组
阅读更多...
redis过期key的删除策略介绍

redis过期key的删除策略介绍

《redis过期key的删除策略介绍》：本文主要介绍redis过期key的删除策略,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录第一种策略：被动删除第二种策略：定期删除第三种策略：强制删除关于big key的清理UNLINK命令FLUSHALL/FLUSHDB命
阅读更多...
C++如何通过Qt反射机制实现数据类序列化

C++如何通过Qt反射机制实现数据类序列化

《C++如何通过Qt反射机制实现数据类序列化》在C++工程中经常需要使用数据类,并对数据类进行存储、打印、调试等操作,所以本文就来聊聊C++如何通过Qt反射机制实现数据类序列化吧... 目录设计预期设计思路代码实现使用方法在 C++ 工程中经常需要使用数据类，并对数据类进行存储、打印、调试等操作。由于数据类
阅读更多...
Python实现图片分割的多种方法总结

Python实现图片分割的多种方法总结

《Python实现图片分割的多种方法总结》图片分割是图像处理中的一个重要任务,它的目标是将图像划分为多个区域或者对象,本文为大家整理了一些常用的分割方法,大家可以根据需求自行选择... 目录1. 基于传统图像处理的分割方法(1) 使用固定阈值分割图片(2) 自适应阈值分割(3) 使用图像边缘检测分割(4)
阅读更多...
Android实现在线预览office文档的示例详解

Android实现在线预览office文档的示例详解

《Android实现在线预览office文档的示例详解》在移动端展示在线Office文档（如Word、Excel、PPT）是一项常见需求,这篇文章为大家重点介绍了两种方案的实现方法,希望对大家有一定的... 目录一、项目概述二、相关技术知识三、实现思路3.1 方案一：WebView + Office Onl
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2