基于netty与springboot实现一对一与群聊天功能

基于netty与springboot实现一对一与群聊天功能
依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>

<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.56</version>
</dependency>

<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.33.Final</version>
</dependency>
</dependencies>

主要yml配置（thymeleaf）
thymeleaf:
cache: false
prefix: classpath:/templates/
suffix: .html
mode: HTML5
encoding: UTF-8
servlet:
content-type: text/html

concurrent包的线程安全Map，用来存放每个客户端对应的channel
import io.netty.channel.Channel;

import java.util.concurrent.ConcurrentHashMap;

public class ChatConfig {
public static ConcurrentHashMap<String, Channel> concurrentHashMap = new ConcurrentHashMap();
}

netty服务端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.sctp.nio.NioSctpServerChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Component
@Slf4j
public class ChatServer {

private EventLoopGroup bossGroup;

private EventLoopGroup workGroup;

private void run(){
log.error(“开始启动聊天服务器”);
bossGroup=new NioEventLoopGroup(1);

workGroup=new NioEventLoopGroup();

try {
ServerBootstrap serverBootstrap =new ServerBootstrap();

serverBootstrap.group(bossGroup,workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChatServerInitializer());
//启动服务器
ChannelFuture channelFuture=serverBootstrap.bind(3002).sync();
log.error(“开始启动聊天服务器完毕….”);
channelFuture.channel().closeFuture().sync();

} catch (Exception e) {
e.printStackTrace();
} finally {

bossGroup.shutdownGracefully();

workGroup.shutdownGracefully();
}

}

/*
* 初始化服务器
* */
@PostConstruct
@SneakyThrows
public void init(){
new Thread(this::run).start();
}

@PreDestroy
public void destroy() throws InterruptedException {
if (bossGroup != null) {
bossGroup.shutdownGracefully().sync();
}
if (workGroup != null) {
workGroup.shutdownGracefully().sync();
}
}
}

初始化配置
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
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 ChatServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline=socketChannel.pipeline();
//使用http的编码器和解码器
pipeline.addLast(new HttpServerCodec());

//添加块处理器主要作用是支持异步发送的码流（大文件传输），但不专用过多的内存，防止java内存溢出
pipeline.addLast(new ChunkedWriteHandler());

//加入ObjectAggregator解码器，作用是他会把多个消息转换为单一的FullHttpRequest或者FullHttpResponse
pipeline.addLast(new HttpObjectAggregator(65536));

// 加入webSocket的hanlder
pipeline.addLast(new WebSocketServerProtocolHandler(“/netty”));

//自定义handler,处理业务逻辑
pipeline.addLast(new ChatServerHandler());
}
}

客户端处理消息
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class ChatServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {

//传过来的是json字符串
String text=textWebSocketFrame.text();
JSONObject jsonObject = JSON.parseObject(text);
//获取到发送人的用户id
Object msg = jsonObject.get(“msg”);
String userId = (String) jsonObject.get(“userId”);
String sendId = (String) jsonObject.get(“sendId”);
Integer type=(Integer)jsonObject.get(“type”);
Channel channel=channelHandlerContext.channel();

log.error(“msg..”+msg);
log.error(“type…”+type);
log.error(“userId…”+userId);
log.error(“sendId…”+sendId);
if (msg==null){
//说明是第一次登录上来连接，还没开始聊天，将uid加到map里
register(userId,channel);
}else {
//有消息了，开始聊天了
if (type==0){
log.error(“单发..”);
sendMsg(msg,userId,sendId);
}else {
log.error(“群发…”);
sendGroupMsg(msg,userId,sendId);
}

}
}

/**
* 第一次登录进来
*
* @param userId
* @param channel
*/
private void register(String userId, Channel channel) {
if (!ChatConfig.concurrentHashMap.containsKey(userId)) { //没有指定的userId
ChatConfig.concurrentHashMap.put(userId, channel);
// 将用户ID作为自定义属性加入到channel中，方便随时channel中获取用户ID
AttributeKey<String> key = AttributeKey.valueOf(“userId”);
channel.attr(key).setIfAbsent(userId);
}
}

/**
* 开发发送消息，进行聊天
*
* @param msg
* @param userId
*/
private void sendMsg(Object msg, String userId,String sendId) {
Channel channel1 = ChatConfig.concurrentHashMap.get(userId);
if (channel1 != null) {
channel1.writeAndFlush(new TextWebSocketFrame(“服务器时间” + LocalDateTime.now() + ” fromId” +sendId+”:”+ msg));
}else {
//存数据库，redis或队列
log.error(“….”);
}
}

/**
* 一旦客户端连接上来，该方法被执行
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
log.info(“handlerAdded 被调用” + ctx.channel().id().asLongText());
}

/**
* 断开连接，需要移除用户
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
removeUserId(ctx);
}

/**
* 移除用户
*
* @param ctx
*/
private void removeUserId(ChannelHandlerContext ctx) {
Channel channel = ctx.channel();
AttributeKey<String> key = AttributeKey.valueOf(“userId”);
String userId = channel.attr(key).get();
ChatConfig.concurrentHashMap.remove(userId);
log.info(“用户下线,userId：{}”, userId);
}

/**
* 处理移除，关闭通道
*
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}

/*
* 群发消息
* */
private void sendGroupMsg(Object msg, String userId,String sendId) {
ConcurrentHashMap<String, Channel> concurrentHashMap = ChatConfig.concurrentHashMap;
for (String s : concurrentHashMap.keySet()) {
sendMsg(msg,s,sendId);
}

}
}

controller层用户登录与发送信息
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;

@Controller
public class ChatController {

@GetMapping(“login”)
public String login(Model model, @RequestParam(“userId”) String userId, @RequestParam(“sendId”) String sendId) {
model.addAttribute(“userId”, userId);
model.addAttribute(“sendId”, sendId);
return “netty”;
}

@GetMapping(“sendMsg”)
public String login(@RequestParam(“sendId”) String sendId) throws InterruptedException {
while (true) {
Channel channel = ChatConfig.concurrentHashMap.get(sendId);
if (channel != null) {
channel.writeAndFlush(new TextWebSocketFrame(“test”));
Thread.sleep(1000);
}
}

}

前端html （netty.html）
<!DOCTYPE html>
<html lang=”en”>
<head>
<meta charset=”UTF-8″>
<title>Title</title><script>
var socket;
//判断当前浏览器是否支持websocket
if (window.WebSocket) {
//go on
socket = new WebSocket(“ws://localhost:3002/netty”);
//相当于channelReado, ev 收到服务器端回送的消息
socket.onmessage = function (ev) {
var rt = document.getElementById(“responseText”);
rt.value = rt.value + “\n” + ev.data;
}

//相当于连接开启(感知到连接开启)
socket.onopen = function (ev) {
var rt = document.getElementById(“responseText”);
rt.value = “连接开启了..”
var userId = document.getElementById(“userId”).value;
var sendId = document.getElementById(“sendId”).value;
var myObj = {userId: userId,sendId:sendId};
var myJSON = JSON.stringify(myObj);
socket.send(myJSON)
}

//相当于连接关闭(感知到连接关闭)
socket.onclose = function (ev) {
var rt = document.getElementById(“responseText”);
rt.value = rt.value + “\n” + “连接关闭了..”
}
} else {
alert(“当前浏览器不支持websocket”)
}

//发送消息到服务器
function send(message) {
if (!window.socket) { //先判断socket是否创建好
return;
}
if (socket.readyState == WebSocket.OPEN) {
//通过socket 发送消息
var userId = document.getElementById(“userId”).value;
var sendId = document.getElementById(“sendId”).value;
// var myObj = {userId: sendId, msg: message};
var type=0;
var myObj = {userId: sendId,sendId:userId,type:type, msg: message};
var messageJson = JSON.stringify(myObj);
socket.send(messageJson)
} else {
alert(“连接没有开启”);
}
}

//群发送消息到服务器
function groupsend(message) {
if (!window.socket) { //先判断socket是否创建好
return;
}
if (socket.readyState == WebSocket.OPEN) {
//通过socket 发送消息
var userId = document.getElementById(“userId”).value;
var sendId = document.getElementById(“sendId”).value;
var type=1 ;
// var myObj = {userId: sendId, msg: message};
var myObj = {userId: sendId,sendId:userId, type:type,msg: message};
var messageJson = JSON.stringify(myObj);
socket.send(messageJson)
} else {
alert(“连接没有开启”);
}
}
</script>
</head>
<body>
<h1 th:text=”${userId}”></h1>
<input type=”hidden” th:value=”${userId}” id=”userId”>
<input type=”hidden” th:value=”${sendId}” id=”sendId”>
<form onsubmit=”return false”>
<textarea name=”message” style=”height: 300px; width: 300px”></textarea>
<input type=”button” value=”发送” onclick=”send(this.form.message.value)”>
<input type=”button” value=”群发送” onclick=”groupsend(this.form.message.value)”>
<textarea id=”responseText” style=”height: 300px; width: 300px”></textarea>
<input type=”button” value=”清空内容” onclick=”document.getElementById(‘responseText’).value=””>
</form>
</body>
</html>
最后浏览器访问如 (http://localhost:3001/login?userId=1&sendId=2)后面的两个id暂为手动添加测试即可

————————————————
版权声明：本文为CSDN博主「橘色月亮」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/qq_37576449/article/details/120431031

基于netty与springboot实现一对一与群聊天功能

相关文章

归档