文章目录

1.简介
2.最终功能实现的效果图
- 2.1.pc端
- 2.2.移动端
3.实战应用
- 3.1.引入依赖
- 3.2.配置文件
- 3.3.测试demo
- - 3.3.1.消息内容实体类
  - 3.3.2.处理请求的Handler类
  - 3.3.3.Netty服务启动类
  - 3.3.4.容器启动后加载Netty服务类
  - 3.3.5.客户端断开连接释放资源
  - 3.3.6.初始化用户群聊信息
  - 3.3.7.页面代码
  - 3.3.3.测试功能
- 3.4.在线客服功能实现
- - 3.4.1.消息内容实体类
  - 3.4.2.处理请求的Handler类
  - 3.4.3.Netty服务启动类
  - 3.4.4.web控制器
  - 3.4.5.测试
4.数据存储
5.项目配套代码

1.简介

Netty
Netty 是一个基于NIO的客户、服务器端的编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。
WebSocket
WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket通信协议于2011年被IETF定为标准RFC 6455，并由RFC7936补充规范。WebSocket API也被W3C定为标准。
WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

2.最终功能实现的效果图

在线客服功能包含2个页面,客服回复页面和游客页面,目前都支持pc端和移动端使用。

2.1.pc端

2.2.移动端

3.实战应用

3.1.引入依赖

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--netty核心组件--><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.36.Final</version></dependency><!--json转换器--><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.14</version></dependency><!--模版引擎--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-thymeleaf</artifactId></dependency>

3.2.配置文件

server:port: 8018netty:#监听websocket连接的端口port: 11111#websocket连接地址  （此处要用电脑的ip,不然手机访问会出现问题）ws: ws://192.168.3.175:${netty.port}/ws

3.3.测试demo

3.3.1.消息内容实体类

public class SocketMessage {/*** 消息类型*/private String messageType;/*** 消息发送者id*/private Integer userId;/*** 消息接受者id或群聊id*/private Integer chatId;/*** 消息内容*/private String message;//....省略get set方法
}

3.3.2.处理请求的Handler类

public class TestWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {private final Logger logger = LoggerFactory.getLogger(TestWebSocketHandler.class);/*** 存储已经登录用户的channel对象*/public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 存储用户id和用户的channelId绑定*/public static ConcurrentHashMap<Integer, ChannelId> userMap = new ConcurrentHashMap<>();/*** 用于存储群聊房间号和群聊成员的channel信息*/public static ConcurrentHashMap<Integer, ChannelGroup> groupMap = new ConcurrentHashMap<>();/*** 获取用户拥有的群聊id号*/UserGroupRepository userGroupRepositor = SpringUtil.getBean(UserGroupRepository.class);@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {logger.info("与客户端建立连接，通道开启！");//添加到channelGroup通道组channelGroup.add(ctx.channel());ctx.channel().id();}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.info("与客户端断开连接，通道关闭！");//添加到channelGroup 通道组channelGroup.remove(ctx.channel());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {//首次连接是FullHttpRequest，把用户id和对应的channel对象存储起来if (null != msg && msg instanceof FullHttpRequest) {FullHttpRequest request = (FullHttpRequest) msg;String uri = request.uri();Integer userId = getUrlParams(uri);userMap.put(getUrlParams(uri), ctx.channel().id());logger.info("登录的用户id是：{}", userId);//第1次登录,需要查询下当前用户是否加入过群,加入过群,则放入对应的群聊里List<Integer> groupIds = userGroupRepositor.findGroupIdByUserId(userId);ChannelGroup cGroup = null;//查询用户拥有的组是否已经创建了for (Integer groupId : groupIds) {cGroup = groupMap.get(groupId);//如果群聊管理对象没有创建if (cGroup == null) {//构建一个channelGroup群聊管理对象然后放入groupMap中cGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);groupMap.put(groupId, cGroup);}//把用户放到群聊管理对象里去cGroup.add(ctx.channel());}//如果url包含参数，需要处理if (uri.contains("?")) {String newUri = uri.substring(0, uri.indexOf("?"));request.setUri(newUri);}} else if (msg instanceof TextWebSocketFrame) {//正常的TEXT消息类型TextWebSocketFrame frame = (TextWebSocketFrame) msg;logger.info("客户端收到服务器数据：{}", frame.text());SocketMessage socketMessage = JSON.parseObject(frame.text(), SocketMessage.class);//处理群聊任务if ("group".equals(socketMessage.getMessageType())) {//推送群聊信息groupMap.get(socketMessage.getChatId()).writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(socketMessage)));} else {//处理私聊的任务，如果对方也在线,则推送消息ChannelId channelId = userMap.get(socketMessage.getChatId());if (channelId != null) {Channel ct = channelGroup.find(channelId);if (ct != null) {ct.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(socketMessage)));}}}}super.channelRead(ctx, msg);}@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {}private static Integer getUrlParams(String url) {if (!url.contains("=")) {return null;}String userId = url.substring(url.indexOf("=") + 1);return Integer.parseInt(userId);}
}

3.3.3.Netty服务启动类

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.channel(NioServerSocketChannel.class)// 绑定监听端口.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 OnlineWebSocketHandler());//添加在线客服聊天消息处理类ch.pipeline().addLast(new TestWebSocketHandler());//添加测试的聊天消息处理类ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));}});// 服务器异步创建绑定ChannelFuture cf = sb.bind().sync();System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());// 关闭服务器通道cf.channel().closeFuture().sync();} finally {// 释放线程池资源group.shutdownGracefully().sync();bossGroup.shutdownGracefully().sync();}}
}

3.3.4.容器启动后加载Netty服务类

@Component
public class NettyInitListen implements CommandLineRunner {@Value("${netty.port}")Integer nettyPort;@Value("${server.port}")Integer serverPort;@Overridepublic void run(String... args) throws Exception {try {System.out.println("nettyServer starting ...");System.out.println("http://127.0.0.1:" + serverPort + "/login");new NettyServer(nettyPort).start();} catch (Exception e) {System.out.println("NettyServerError:" + e.getMessage());}}
}

3.3.5.客户端断开连接释放资源

/*** @author Dominick Li* @createTime 2020/3/8  16:07* @description session超时, 移除 websocket对应的channel**/
public class MySessionListener implements HttpSessionListener {private final Logger logger = LoggerFactory.getLogger(MySessionListener.class);@Overridepublic void sessionCreated(HttpSessionEvent httpSessionEvent) {logger.info("sessionCreated sessionId={}", httpSessionEvent.getSession().getId());MySessionContext.AddSession(httpSessionEvent.getSession());}@Overridepublic void sessionDestroyed(HttpSessionEvent httpSessionEvent) {HttpSession session = httpSessionEvent.getSession();Integer userId = session.getAttribute("userId") == null ? null : Integer.parseInt(session.getAttribute("userId").toString());//销毁时重websocket channel中移除if (userId != null) {ChannelId channelId = TestWebSocketHandler.userMap.get(userId);if (channelId != null) {//移除了私聊的channel对象, 群聊的还未移除TestWebSocketHandler.userMap.remove(userId);TestWebSocketHandler.channelGroup.remove(channelId);logger.info("session timeout,remove channel, userId={}", userId);}}MySessionContext.DelSession(session);logger.info("session destroyed  .... ");}public static class MySessionContext {private static HashMap mymap = new HashMap();public static synchronized void AddSession(HttpSession session) {if (session != null) {mymap.put(session.getId(), session);}}public static synchronized void DelSession(HttpSession session) {if (session != null) {mymap.remove(session.getId());}}public static synchronized HttpSession getSession(String session_id) {if (session_id == null) {return null;}return (HttpSession) mymap.get(session_id);}}
}

3.3.6.初始化用户群聊信息

public interface UserGroupRepository {List<Integer> findGroupIdByUserId(Integer userId);
}
@Component
public class UserGroupRepositoryImpl implements UserGroupRepository {/*** 组装假数据,真实环境应该重数据库获取*/HashMap<Integer, List<Integer>> userGroup = new HashMap<>(4);{List<Integer> list = Arrays.asList(1, 2);userGroup.put(1, list);userGroup.put(2, list);userGroup.put(3, list);userGroup.put(4, list);}@Overridepublic List<Integer> findGroupIdByUserId(Integer userId) {return this.userGroup.get(userId);}
}
```·**web控制器层**
@Controller
public class TestController {@Value("${netty.ws}")private String ws;@AutowiredUserGroupRepository userGroupRepository;/*** 登录页面*/@RequestMapping("/login")public String login() {return "test/login";}/*** 登录后跳转到测试主页*/@PostMapping("/login.do")public String login(@RequestParam Integer userId, HttpSession session, Model model) {model.addAttribute("ws", ws);session.setAttribute("userId", userId);model.addAttribute("groupList", userGroupRepository.findGroupIdByUserId(userId));return "test/index";}
}

3.3.7.页面代码

登录页面

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>Title</title>
</head>
<body>
<form action="/login.do" method="post">登录(默认的4个用户id:[1,2,3,4])用户Id:<input type="number" name="userId"/><input type="submit" value="登录"/>
</form>
</body>
</html>

测试主页

<!DOCTYPE html>
<html lang="en" xmlns:th="http://www.thymeleaf.org">
<head><meta charset="UTF-8"><title>Title</title>
</head>
<style type="text/css">.flexBox {display: flex;width: 100%;}.flexBox div {width: 50%;background-color: pink;}#messageBox ul {border: solid 1px #ccc;width: 600px;height: 400px}
</style>
<body><div class="flexBox"><div style="text-align: right;" th:text="'当前登录的用户:'+${session.userId}"></div>
</div>
<!-- 聊天息 -->
<div class="flexBox" id="messageBox"><ul th:id="${groupId}" th:each="groupId,iterObj : ${groupList}"><li th:text="房间号+${groupId}"></li></ul><ul id="chat"><li>好友消息</li></ul>
</div>
<div style="width:100%;border: solid 1px #ccc;"><form style="width: 40%;border: solid 1px red;margin: 0px auto"><h3>给好友或者群聊发送数据</h3><div>测试数据: (好友 1-4 ,房间号 1-2 )<br/>请输出好友编号或房间号 <input type="number" id="chatId" value="1"><br/><textarea id="message" style="width: 100%">在不?</textarea></div><div>消息类型<input name="messageType" type="radio" checked value="group">群聊<input name="messageType" type="radio" value="chat">私聊<a href="#" id="send">发送</a></div></form>
</div>
</body>
<!--在js脚本中获取作用域的值-->
<script th:inline="javascript">//获取session中的uservar userId = [[${session.userId}]];//获取ws服务地址var ws = [[${ws}]]
</script><script type="text/javascript">var websocket;if (!window.WebSocket) {window.WebSocket = window.MozWebSocket;}if (window.WebSocket) {websocket = new WebSocket(ws + "?userId=" + userId);websocket.onmessage = function (event) {var json = JSON.parse(event.data);console.log(json)chat.onmessage(json);};websocket.onopen = function (event) {console.log("Netty-WebSocket服务器。。。。。。连接");};websocket.onclose = function (event) {console.log("Netty-WebSocket服务器。。。。。。关闭");};} else {alert("您的浏览器不支持WebSocket协议！");}//监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。window.onbeforeunload = function () {if (websocket != null) {websocket.close();}};
</script><script>/*** sendMessage    发送消息推送给websocket对象* onmessage      接受来自服务端推送的消息,并显示在页面* */var chat = {sendMessage: function () {var message = document.getElementById("message").value; //发送的内容if (message == "") {alert('不能发送空消息');return;}if (!window.WebSocket) {return;}var chatId = document.getElementById("chatId").value; //好友Id或房间号idvar radio=document.getElementsByName("messageType");var messageType=null;   //  聊天类型for(var i=0;i<radio.length;i++){if(radio[i].checked==true) {messageType=radio[i].value;break;}}if (messageType == "chat") {if (chatId == userId) {alert("不能给自己发私聊信息,请换个好友吧");}var li = document.createElement("li");li.innerHTML = "My:" + messagevar ul = document.getElementById("chat");ul.appendChild(li);}if (websocket.readyState == WebSocket.OPEN) {var data = {};data.chatId = chatId;data.message = message;data.userId = userId;data.messageType = messageType;websocket.send(JSON.stringify(data));} else {alert("和服务器连接异常！");}},onmessage: function (jsonData) {var id;if (jsonData.messageType == "chat") {id = "chat";} else {id = jsonData.chatId;}console.log(id);var li = document.createElement("li");li.innerHTML = "用户id=" + jsonData.userId + ":" + jsonData.message;var ul = document.getElementById(id);ul.appendChild(li);}}document.onkeydown = keyDownSearch;function keyDownSearch(e) {// 兼容FF和IE和Opera    var theEvent = e || window.event;var code = theEvent.keyCode || theEvent.which || theEvent.charCode;// 13 代表 回车键if (code == 13) {// 要执行的函数 或者点击事件chat.sendMessage();return false;}return true;}document.getElementById("send").onclick = function () {chat.sendMessage();}
</script>
</html>

3.3.3.测试功能

访问登录页面: http://localhost:8018/login
分别打开2个浏览器,一个用 id=1登录,另外一个用id=2登录

选择消息类型为群聊,然后输入房间号就可以发送群聊消息了

选择消息类型为私聊,然后输入好友Id就可以发送私聊信息

3.4.在线客服功能实现

3.4.1.消息内容实体类

public class OnlineMessage {/*** 消息发送者id*/private String sendId;/*** 消息接受者id*/private String acceptId;/*** 消息内容*/private String message;/*** 头像*/private String headImg;//省略get set方法
}

3.4.2.处理请求的Handler类

public class OnlineWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {private final Logger logger = LoggerFactory.getLogger(TestWebSocketHandler.class);/*** 存储已经登录用户的channel对象*/public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 存储用户id和用户的channelId绑定*/public static ConcurrentHashMap<String, ChannelId> userMap = new ConcurrentHashMap<>();@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {logger.info("与客户端建立连接，通道开启！");//添加到channelGroup通道组channelGroup.add(ctx.channel());//ctx.channel().id();}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.info("与客户端断开连接，通道关闭！");//添加到channelGroup 通道组channelGroup.remove(ctx.channel());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {//首次连接是FullHttpRequest，把用户id和对应的channel对象存储起来if (null != msg && msg instanceof FullHttpRequest) {FullHttpRequest request = (FullHttpRequest) msg;String uri = request.uri();String userId = getUrlParams(uri);//登录后把用户id和channel关联上userMap.put(userId, ctx.channel().id());logger.info("登录的用户id是：{}", userId);//如果url包含参数，需要处理if (uri.contains("?")) {String newUri = uri.substring(0, uri.indexOf("?"));request.setUri(newUri);}} else if (msg instanceof TextWebSocketFrame) {//正常的TEXT消息类型TextWebSocketFrame frame = (TextWebSocketFrame) msg;logger.info("客户端收到服务器数据：{}", frame.text());OnlineMessage onlineMessage = JSON.parseObject(frame.text(), OnlineMessage.class);//处理私聊的任务，如果对方也在线,则推送消息ChannelId channelId = userMap.get(onlineMessage.getAcceptId());if (channelId != null) {Channel ct = channelGroup.find(channelId);if (ct != null) {ct.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(onlineMessage)));}}}super.channelRead(ctx, msg);}@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {}/*** 解析url中的参数* @return 获取用户的id*/private String getUrlParams(String url) {if (!url.contains("=")) {return null;}String userId = url.substring(url.indexOf("=") + 1);return userId;}
}

3.4.3.Netty服务启动类

和测试demo用的一致,只是handler处理类不一致,把测试的注释掉,在线聊天的handler放开,加载netty服务类用同一个即可

3.4.4.web控制器

@Controller
public class OnlineController {@Value("${netty.ws}")private String ws;/*** 客服界面*/@GetMapping(value = {"/index", "/customer","/"})public String index(Model model) {model.addAttribute("ws", ws);return "customer";}/*** 游客页面*/@GetMapping("/tourist")public String tourist(Model model) {model.addAttribute("ws", ws);return "tourist";}
}

3.4.5.测试

在线客户的html代码和js代码请参考本博客配套代码

访问游客页面: http://localhost:8018/tourist

访问客服页面: http://localhost:8018/customer

4.数据存储

该项目是demo级别的,没有实现聊天记录持久化，如需持久化,则要自己实现聊天数据存储到数据库或者redis里面,建议存放到redis中,存在redis可以方便设置消息过期自动清理的天数。

5.项目配套代码

github地址

创作不易,要是觉得我写的对你有点帮助的话,麻烦在github上帮我点下 Star

【SpringBoot框架篇】其它文章如下,后续会继续更新。

1.搭建第一个springboot项目
2.Thymeleaf模板引擎实战
3.优化代码,让代码更简洁高效
4.集成jta-atomikos实现分布式事务
5.分布式锁的实现方式
6.docker部署,并挂载配置文件到宿主机上面
7.项目发布到生产环境
8.搭建自己的spring-boot-starter
9.dubbo入门实战
10.API接口限流实战
11.Spring Data Jpa实战
12.使用druid的monitor工具查看sql执行性能
13.使用springboot admin对springboot应用进行监控
14.mybatis-plus实战
15.使用shiro对web应用进行权限认证
16.security整合jwt实现对前后端分离的项目进行权限认证
17.使用swagger2生成RESTful风格的接口文档
18.使用Netty加websocket实现在线聊天功能
19.使用spring-session加redis来实现session共享
20.自定义@Configuration配置类启用开关
21.对springboot框架编译后的jar文件瘦身
22.集成RocketMQ实现消息发布和订阅
23.集成smart-doc插件零侵入自动生成RESTful格式API文档
24.集成FastDFS实现文件的分布式存储
25.集成Minio实现文件的私有化对象存储
26.集成spring-boot-starter-validation对接口参数校验
27.集成mail实现邮件推送带网页样式的消息
28.使用JdbcTemplate操作数据库
29.Jpa+vue实现单模型的低代码平台
30.使用sharding-jdbc实现读写分离和分库分表
31.基于分布式锁或xxl-job实现分布式任务调度
32.基于注解+redis实现表单防重复提交

【SpringBoot框架篇】18.使用Netty加websocket实现在线聊天功能相关推荐

JavaWeb--使用Websocket实现在线聊天功能
首先简单介绍下WebSocket,WebSocket是HTML5中内容,是基于TCP的一种新的网络协议,它支持全双工.长连接的通信.在它出现之前,实时消息发送与接收通过轮询实现,但是频繁与服务器建立连 ...
【SpringBoot框架篇】11.Spring Data Jpa实战
文章目录 1.简介 1.1.JPA 1.2.Spring Data Jpa 1.3.Hibernate 1.4.Jpa.Spring Data Jpa.Hibernate三者之间的关系 2.引入依赖 ...
【SpringBoot框架篇】31.基于分布式锁或xxx-job实现分布式任务调度
文章目录 1.简介 2.分布式锁实现 2.1.引用依赖 2.2.定义分布式锁注解 2.3.配置切入点和获取锁释放锁逻辑 2.4.测试任务 3.使用分布式任务调度平台xxx-job 3.1.下载源码并运 ...
【SpringBoot框架篇】4.集成jta-atomikos实现分布式事务
文章目录 1.简介 1.1.分布式事务 1.2.JTA 1.3.atomikos 2.引入依赖 3.修改配置文件 4.添加配置读取类 5.创建多数据源,管理事务 6.数据持久化 7.数据库DLL文件 ...
Springboot + WebSocket 实现在线聊天
一.后端 1.在Springboot项目的pom.xml中添加依赖  <dependency><groupId>org.spri ...
springboot+websocket构建在线聊天室（群聊+单聊）
系列导读: 1.springboot+websocket构建在线聊天室(群聊+单聊) 2.Spring Boot WebSocket:单聊(实现思路) 3.Websocket Stomp+Rabbit ...
springboot -- 整合websocket 实现在线聊天
项目demo: 链接:https://pan.baidu.com/s/1xZtU-Rqc58m0-v397OW3hQ 提取码:01pt 复制这段内容后打开百度网盘手机App,操作更方便哦 websoc ...
SpringBoot与webSocket实现在线聊天室——实现私聊+群聊+聊天记录保存
SpringBoot与webSocket实现在线聊天室--实现私聊+群聊+聊天记录保存引用参考:原文章地址:https://blog.csdn.net/qq_41463655/article/det ...
WebSocket实现在线聊天
WebSocket实现在线聊天前两天在公司接到一个需求,使用WebSocket实现微信扫码登陆,当时了解了一下WebSocket,都说WebSocket可以实现在线聊天,所以我自己也写了一个.(发个 ...

【SpringBoot框架篇】18.使用Netty加websocket实现在线聊天功能