Springboot+Netty+WebSocket搭建简单的消息通知

Springboot+Netty+WebSocket搭建简单的消息通知

一、快速开始

1、添加依赖
<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.36.Final</version>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
2、添加配置

spring:http:encoding:force: truecharset: UTF-8application:name: spring-cloud-study-websocket-reidsfreemarker:request-context-attribute: request#prefix: /templates/suffix: .htmlcontent-type: text/htmlenabled: truecache: falsecharset: UTF-8allow-request-override: falseexpose-request-attributes: trueexpose-session-attributes: trueexpose-spring-macro-helpers: true#template-loader-path: classpath:/templates/
3、添加启动类
@SpringBootApplication
public class WebSocketApplication {public static void main(String[] args) {SpringApplication.run(WebSocketApplication.class);try {new NettyServer(12345).start();System.out.println("https://blog.csdn.net/moshowgame");System.out.println("http://127.0.0.1:6688/netty-websocket/index");}catch(Exception e) {System.out.println("NettyServerError:"+e.getMessage());}}
}

二、添加WebSocket部分代码

1、WebSocketServer

@Slf4j
@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketServer {/**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/private static AtomicInteger onlineCount = new AtomicInteger(0);/**concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。*/private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();/**与某个客户端的连接会话，需要通过它来给客户端发送数据*/private Session session;/**接收userId*/private String userId="";/*** 连接建立成功调用的方法*/@OnOpenpublic void onOpen(Session session,@PathParam("userId") String userId) {this.session = session;this.userId=userId;if(webSocketMap.containsKey(userId)){webSocketMap.remove(userId);webSocketMap.put(userId,this);}else{webSocketMap.put(userId,this);onlineCount.incrementAndGet(); // 在线数加1}log.info("用户连接:"+userId+",当前在线人数为:" +  onlineCount.get());try {sendMessage("连接成功");} catch (IOException e) {log.error("用户:"+userId+",网络异常!!!!!!");}}/*** 连接关闭调用的方法*/@OnClosepublic void onClose() {if(webSocketMap.containsKey(userId)){webSocketMap.remove(userId);onlineCount.decrementAndGet(); // 在线数减1}log.info("用户退出:"+userId+",当前在线人数为:" +  onlineCount.get());}/*** 收到客户端消息后调用的方法** @param message 客户端发送过来的消息*/@OnMessagepublic void onMessage(String message, Session session) {log.info("用户消息:"+userId+",报文:"+message);//可以群发消息//消息保存到数据库、redisif(StrUtil.isNotBlank(message)){try {//解析发送的报文JSONObject jsonObject = JSON.parseObject(message);//追加发送人(防止串改)jsonObject.put("fromUserId",this.userId);String toUserId=jsonObject.getString("toUserId");//传送给对应toUserId用户的websocketif(StrUtil.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());}else{log.error("请求的userId:"+toUserId+"不在该服务器上");//否则不在这个服务器上，发送到mysql或者redis}}catch (Exception e){e.printStackTrace();}}}/**** @param session* @param error*/@OnErrorpublic void onError(Session session, Throwable error) {log.error("用户错误:"+this.userId+",原因:"+error.getMessage());error.printStackTrace();}/*** 实现服务器主动推送*/public void sendMessage(String message) throws IOException {this.session.getBasicRemote().sendText(message);}/*** 发送自定义消息* */public static void sendInfo(String message,@PathParam("userId") String userId) throws IOException {log.info("发送消息到:"+userId+"，报文:"+message);if(StrUtil.isNotBlank(userId)&&webSocketMap.containsKey(userId)){webSocketMap.get(userId).sendMessage(message);}else{log.error("用户"+userId+",不在线！");}}public static synchronized AtomicInteger getOnlineCount() {return onlineCount;}
}
2、WebSocketConfig
@Configuration
public class WebSocketConfig {  @Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();  }  
}
3、DemoController
import cn.vipthink.socket.server.WebSocketServer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;import java.io.IOException;@RestController
public class DemoController {@GetMapping("index")public ResponseEntity<String> index(){return ResponseEntity.ok("请求成功");}@GetMapping("page")public ModelAndView page(){return new ModelAndView("index");}@RequestMapping("/push/{toUserId}")public ResponseEntity<String> pushToWeb(String message, @PathVariable String toUserId) throws IOException {WebSocketServer.sendInfo(message,toUserId);return ResponseEntity.ok("MSG SEND SUCCESS");}
}
6、添加templates/index.html
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head><meta http-equiv="Content-Type" content="text/html; charset=utf-8" /><title>Netty-Websocket</title><script type="text/javascript">// by zhengkai.blog.csdn.netvar socket;if(!window.WebSocket){window.WebSocket = window.MozWebSocket;}if(window.WebSocket){socket = new WebSocket("ws://127.0.0.1:12345/ws");socket.onmessage = function(event){var ta = document.getElementById('responseText');ta.value += event.data+"\r\n";};socket.onopen = function(event){var ta = document.getElementById('responseText');ta.value = "Netty-WebSocket服务器。。。。。。连接  \r\n";};socket.onclose = function(event){var ta = document.getElementById('responseText');ta.value = "Netty-WebSocket服务器。。。。。。关闭 \r\n";};}else{alert("您的浏览器不支持WebSocket协议！");}function send(message){if(!window.WebSocket){return;}if(socket.readyState == WebSocket.OPEN){socket.send(message);}else{alert("WebSocket 连接没有建立成功！");}}</script>
</head>
<body>
<form onSubmit="return false;"><label>ID</label><input type="text" name="uid" value="${uid!!}" /> <br /><label>TEXT</label><input type="text" name="message" value="这里输入消息" /> <br /><br /> <input type="button" value="发送ws消息"onClick="send(this.form.uid.value+':'+this.form.message.value)" /><hr color="black" /><h3>服务端返回的应答消息</h3><textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
</form>
</body>
</html>

三、添加Netty部分

1、NettyServer
import cn.vipthink.socket.handler.WSWebSocketHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;public class NettyServer {private final int port;public NettyServer(int port) {this.port = port;}public void start() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup group = new NioEventLoopGroup();try {ServerBootstrap sb = new ServerBootstrap();sb.option(ChannelOption.SO_BACKLOG, 1024);sb.group(group, bossGroup) // 绑定线程池.channel(NioServerSocketChannel.class) // 指定使用的channel.localAddress(this.port)// 绑定监听端口.childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作@Overrideprotected void initChannel(SocketChannel ch) throws Exception {System.out.println("收到新连接");//websocket协议本身是基于http协议的，所以这边也要使用http解编码器ch.pipeline().addLast(new HttpServerCodec());//以块的方式来写的处理器ch.pipeline().addLast(new ChunkedWriteHandler());ch.pipeline().addLast(new HttpObjectAggregator(8192));ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));ch.pipeline().addLast(new WSWebSocketHandler());}});ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());cf.channel().closeFuture().sync(); // 关闭服务器通道} finally {group.shutdownGracefully().sync(); // 释放线程池资源bossGroup.shutdownGracefully().sync();}}
}
2、WSChannelHandlerPool
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;/*** WSChannelHandlerPool* 通道组池，管理所有websocket连接*/
public class WSChannelHandlerPool {public WSChannelHandlerPool(){}public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);}
3、WSWebSocketHandler
import cn.vipthink.socket.config.WSChannelHandlerPool;
import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;import java.util.HashMap;
import java.util.Map;public class WSWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("与客户端建立连接，通道开启！");//添加到channelGroup通道组WSChannelHandlerPool.channelGroup.add(ctx.channel());}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println("与客户端断开连接，通道关闭！");//添加到channelGroup 通道组WSChannelHandlerPool.channelGroup.remove(ctx.channel());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {//首次连接是FullHttpRequest，处理参数 by zhengkai.blog.csdn.netif (null != msg && msg instanceof FullHttpRequest) {FullHttpRequest request = (FullHttpRequest) msg;String uri = request.uri();Map paramMap=getUrlParams(uri);System.out.println("接收到的参数是："+ JSON.toJSONString(paramMap));//如果url包含参数，需要处理if(uri.contains("?")){String newUri=uri.substring(0,uri.indexOf("?"));System.out.println(newUri);request.setUri(newUri);}}else if(msg instanceof TextWebSocketFrame){//正常的TEXT消息类型TextWebSocketFrame frame=(TextWebSocketFrame)msg;System.out.println("客户端收到服务器数据：" +frame.text());sendAllMessage(frame.text());}super.channelRead(ctx, msg);}@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {}private void sendAllMessage(String message){//收到信息后，群发给所有channelWSChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));}private static Map getUrlParams(String url){Map<String,String> map = new HashMap<>();url = url.replace("?",";");if (!url.contains(";")){return map;}if (url.split(";").length > 0){String[] arr = url.split(";")[1].split("&");for (String s : arr){String key = s.split("=")[0];String value = s.split("=")[1];map.put(key,value);}return  map;}else{return map;}}
}

四、启动服务

http://localhost:9999/demo/page

可以通过前端发送消息到后端，通过日志查看

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/49267.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

c++ 友元 运算符重载详解

c++ 友元 运算符重载详解

友元 c是面向对象的&#xff0c;目的之一&#xff1a;封装 封装&#xff1a; 优点之一&#xff0c;就是安全。 缺点&#xff1a;在某些特殊的场合&#xff0c;不是很方便。 华为与IBM 40亿的咨询故事 IBM需要对华为各级部门做深度咨询分析&#xff0c; 为了提高咨询效率&a…
阅读更多...
Transformer在医学影像中的应用综述-分类

Transformer在医学影像中的应用综述-分类

文章目录 COVID-19 Diagnosis黑盒模型可解释的模型 肿瘤分类黑盒模型可解释模型 视网膜疾病分类小结 总体结构 COVID-19 Diagnosis 黑盒模型 Point-of-Care Transformer(POCFormer)&#xff1a;利用Linformer将自注意的空间和时间复杂度从二次型降低到线性型。POCFormer有200…
阅读更多...
Docker基础操作

Docker基础操作

1.安装docker服务&#xff0c;配置镜像加速器 安装docker服务 清理缓存 sudo yum remove docker \ docker-client \ docker-client-latest \ docker-common \ docker-latest \ docker-latest-logrotate \ docker-logrotate \ docker-enginesystemctl enable --now docker 脚…
阅读更多...
Vue3+Vite 初始化Cesium

Vue3+Vite 初始化Cesium

Vue3Vite 初始化Cesium 安装依赖 yarn add cesium yarn add vite-plugin-cesium -D加载vite-plugin-cesium插件 import { defineConfig } from vite import vue from vitejs/plugin-vue import cesium from vite-plugin-cesium;export default defineConfig({plugins: [vue(…
阅读更多...
微积分进阶 1.1 函数

微积分进阶 1.1 函数

一、函数的概念 在观察自然现象或工程实际问题时&#xff0c;我们经常发现有几个变量在变化&#xff0c;这些变量之间并不是彼此孤立的&#xff0c;而是相互制约的&#xff0c;这些变量是怎么变化的呢&#xff1f;它们之间有什么联系呢&#xff1f;存什么规律呢&#xff1f;怎…
阅读更多...
WPF中手写地图控件（4）——离线地图

WPF中手写地图控件（4）——离线地图

内存缓存和本地文件缓存技术 如果每个瓦片图每次打开都要重新加载&#xff0c;会比较浪费资源&#xff0c;而且如果网络不好&#xff0c;甚至要等很久&#xff0c;于是可以使用内存缓存&#xff0c;每次已经加载的瓦片图&#xff0c;第二次再加载之前&#xff0c;先看看内存中…
阅读更多...
ARM开发（cortex-A7核中断实验)

ARM开发（cortex-A7核中断实验)

1.实验目的&#xff1a;实现KEY1/LEY2/KE3三个按键&#xff0c;中断触发打印一句话&#xff0c;并且灯的状态取反&#xff1b; key1 ----> LED3灯状态取反&#xff1b; key2 ----> LED2灯状态取反&#xff1b; key3 ----> LED1灯状态取反&#xff1b; 2.分析框图: …
阅读更多...
IDEA快速设置Services窗口

IDEA快速设置Services窗口

现在微服务下面会有很多SpringBoot服务&#xff0c;Services窗口方便我们管理各个SpringBoot服务&#xff0c;但有时IDEA打开项目后无法的看到Services窗口&#xff0c;以下步骤可以解决&#xff01;
阅读更多...
leetcode 151. 反转字符串中的单词

leetcode 151. 反转字符串中的单词

反转字符串中的单词 给你一个字符串 s &#xff0c;请你反转字符串中 单词 的顺序。 单词 是由非空格字符组成的字符串。s 中使用至少一个空格将字符串中的 单词 分隔开。 返回 单词 顺序颠倒且 单词 之间用单个空格连接的结果字符串。 注意&#xff1a;输入字符串 s中可能…
阅读更多...
Windows环境使用bat脚本启动Java服务

Windows环境使用bat脚本启动Java服务

Java项目一般会被打包成jar后启动&#xff0c;在windows系统中可以通过终端窗口cmd启动jar包&#xff0c;即在jar包所在的目录中打开cmd&#xff0c;或在cmd中进入到jar包目录&#xff0c;执行如下命令&#xff1a; java -jar test.jar 在bat脚本中执行java服务&#xff0c;命…
阅读更多...
西瓜书之神经网络

西瓜书之神经网络

一&#xff0c;神经元模型 所谓神经网络&#xff0c; 目前用得最广泛的一个定义是“神经网络是由具有适应性的简单单元组成的广泛并行互连的网络&#xff0c;它的组织能够模拟生物神经系统对真实世界物体所做出的交互反应”。 M-P神经元 M-P神经元&#xff1a;接收n个输入(…
阅读更多...
Nodejs-nrm：快速切换npm源 / npm官方源和其他自定义源之间切换

Nodejs-nrm：快速切换npm源 / npm官方源和其他自定义源之间切换

一、理解 Nodejs nrm Nodejs nrm 是一个管理 npm 源的工具。由于 npm 在国内的速度较慢&#xff0c;很多开发者会使用淘宝的 npm 镜像源&#xff0c;但是也会遇到一些问题&#xff0c;例如某些包在淘宝镜像源中不存在&#xff0c;或者淘宝镜像源本身也会有问题。 Nodejs nrm …
阅读更多...
计算机竞赛 基于CNN实现谣言检测 - python 深度学习 机器学习

计算机竞赛 基于CNN实现谣言检测 - python 深度学习 机器学习

文章目录 1 前言1.1 背景 2 数据集3 实现过程4 CNN网络实现5 模型训练部分6 模型评估7 预测结果8 最后 1 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 基于CNN实现谣言检测 该项目较为新颖&#xff0c;适合作为竞赛课题方向&#xff0c;学长非常推荐&am…
阅读更多...
JavaScript中的设计模式之一--单例模式和模块

JavaScript中的设计模式之一--单例模式和模块

虽然有一种疯狂天才的感觉可能很诱人&#xff0c;但重新发明轮子通常不是设计软件的最佳方法。很有可能有人已经遇到了和你一样的问题&#xff0c;并以一种聪明的方式解决了它。这样的最佳实践在形式化后被称为设计模式。今天我们来看看它们的概念&#xff0c;并检查单例模式和…
阅读更多...
AWS解决方案日：Web 3业务安全方案

AWS解决方案日：Web 3业务安全方案

近日&#xff0c;AWS合作伙伴之Web3解决方案日在香港举办&#xff0c;多家科技公司专家和企业代表就WEB 3.0方案、AI创新和Web 3.0安全进行了探讨。顶象现场展示了Web 3.0业务安全解决方案。 NFT是Web 3.0典型场景之一。NFT基于区块链技术的非同质化代币&#xff0c;具有不可分…
阅读更多...
Java入坑之 数据库编程

Java入坑之 数据库编程

一、基础概念 1.1JDBC 步骤 导入驱动jar包 注册驱动 获取数据库连接对象 Connection DataSource dSource; dSource.getConnection(); 定义sql语句 String sql "update account set balance 500 where id 1"; 获取执行sql语句的对象 Statement PreparedStatement…
阅读更多...
在mac下，使用Docker安装达梦数据库

在mac下，使用Docker安装达梦数据库

前言&#xff1a;因为业务需要安装达梦数据库 获取官网下载tar包&#xff08;达梦官网的下载页面https://www.dameng.com/list_103.html&#xff09;&#xff0c;或者通过命令 一、下载tar包 命令下载&#xff1a;wget -O dm8_docker.tar -c https://download.dameng.com/eco/…
阅读更多...
实战：大数据Spark简介与docker-compose搭建独立集群

实战：大数据Spark简介与docker-compose搭建独立集群

文章目录 前言技术积累Spark简介Spark核心功能及优势Spark运行架构 Spark独立集群搭建安装docker和docker-composedocker-compose编排docker-compose编排并运行容器 Spark集群官方案例测试写在最后 前言 很多同学都使用过经典的大数据分布式计算框架hadoop&#xff0c;其分布式…
阅读更多...
学Python静不下来，看了一堆资料还是很迷茫是为什么

学Python静不下来，看了一堆资料还是很迷茫是为什么

一、前言 最近发现&#xff0c;身边很多的小伙伴学Python都会遇到一个问题&#xff0c;就是资料也看了很多&#xff0c;也花了很多时间去学习但还是很迷茫&#xff0c;时间长了又发现之前学的知识点很多都忘了&#xff0c;都萌生出了想半路放弃的想法。 让我们看看蚂蚁金服的大…
阅读更多...
Apache zookeeper kafka 开启SASL安全认证 —— 筑梦之路

Apache zookeeper kafka 开启SASL安全认证 —— 筑梦之路

简介 Kafka是一个高吞吐量、分布式的发布-订阅消息系统。Kafka核心模块使用Scala语言开发&#xff0c;支持多语言&#xff08;如Java、Python、Go等&#xff09;客户端&#xff0c;它可以水平扩展和具有高吞吐量特性而被广泛使用&#xff0c;并与多类开源分布式处理系统进行集成…
阅读更多...
最新文章

Copyright @ 2022~2023