经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Spring » 查看文章
基于SpringBoot+Netty实现即时通讯（IM）功能
来源：cnblogs　　作者：seizedays　　时间：2023/10/20 8:45:58　　对本文有异议

简单记录一下实现的整体框架，具体细节在实际生产中再细化就可以了。

第一步 引入netty依赖

SpringBoot的其他必要的依赖像Mybatis、Lombok这些都是老生常谈了 就不在这里放了

  1.        <dependency>
  2.             <groupId>io.netty</groupId>
  3.             <artifactId>netty-all</artifactId>
  4.             <version>4.1.85.Final</version>
  5.         </dependency>

 

第二步 接下来就是准备工作。

消息服务类（核心代码） 聊天服务的功能就是靠这个类的start()函数来启动的 绑定端口8087 之后可以通socket协议访问这个端口来执行通讯

  1. import com.bxt.demo.im.handler.WebSocketHandler;
  2. import io.netty.bootstrap.ServerBootstrap;
  3. import io.netty.channel.*;
  4. import io.netty.channel.group.ChannelGroup;
  5. import io.netty.channel.group.DefaultChannelGroup;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.codec.http.HttpObjectAggregator;
  10. import io.netty.handler.codec.http.HttpServerCodec;
  11. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
  12. import io.netty.handler.stream.ChunkedWriteHandler;
  13. import io.netty.util.concurrent.GlobalEventExecutor;
  14. import lombok.extern.slf4j.Slf4j;
  16. import java.util.Map;
  17. import java.util.concurrent.ConcurrentHashMap;
  19. /**
  20.  * @Description: 即时通讯服务类
  21.  * @author: bhw
  22.  * @date: 2023年09月27日 13:44
  23.  */
  24. @Slf4j
  25. public class IMServer {
    　　//  用来存放连入服务器的用户集合
  26.     public static final Map<String, Channel> USERS = new ConcurrentHashMap<>(1024);
  27. 　　//  用来存放创建的群聊连接
  28.     public static final ChannelGroup GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  30.     public static void start() throws InterruptedException {
  31.         log.info("IM服务开始启动");
  32.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  33.         EventLoopGroup workGroup = new NioEventLoopGroup();
  35.         // 绑定端口
  36.         ServerBootstrap bootstrap = new ServerBootstrap();
  37.         bootstrap.group(bossGroup,workGroup)
  38.                 .channel(NioServerSocketChannel.class)
  39.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  40.                     @Override
  41.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
  42.                         ChannelPipeline pipeline = socketChannel.pipeline();
  43.                         // 添加http编码解码器
  44.                         pipeline.addLast(new HttpServerCodec())
  45.                                 //支持大数据流
  46.                                 .addLast(new ChunkedWriteHandler())
  47.                                 // 对http消息做聚合操作 FullHttpRequest FullHttpResponse
  48.                                 .addLast(new HttpObjectAggregator(1024*64))
  49.                                 //支持websocket
  50.                                 .addLast(new WebSocketServerProtocolHandler("/"))
  51.                                 .addLast(new WebSocketHandler());
  52.                     }
  53.                 });
  55.         ChannelFuture future = bootstrap.bind(8087).sync();
  56.         log.info("服务器启动开始监听端口: {}", 8087);
  57.         future.channel().closeFuture().sync();
  58.         //关闭主线程组
  59.         bossGroup.shutdownGracefully();
  60.         //关闭工作线程组
  61.         workGroup.shutdownGracefully();
  62.     }
  64. }

 

 创建聊天消息实体类

  1. /**
  2.  * @Description: 聊天消息对象 可以自行根据实际业务扩展
  3.  * @author: seizedays
  4.  */
  5. @Data
  6. public class ChatMessage extends IMCommand {
  7.     //消息类型
  8.     private Integer type;
  9.     //消息目标对象
  10.     private String target;
  11.     //消息内容
  12.     private String content;
  14. }

连接类型枚举类，暂时定义为建立连接、发送消息和加入群组三种状态码

  1. @AllArgsConstructor
  2. @Getter
  3. public enum CommandType {
  5.     //建立连接
  6.     CONNECT(10001),
  7.     //发送消息
  8.     CHAT(10002),
  9.     //加入群聊
  10.     JOIN_GROUP(10003),
  11.     ERROR(-1)
  12.     ;
  15.     private Integer code;
  17.     public static CommandType match(Integer code){
  18.         for (CommandType value : CommandType.values()) {
  19.             if (value.code.equals(code)){
  20.                 return value;
  21.             }
  22.         }
  23.         return ERROR;
  24.     }
  26. }

命令动作为聊天的时候 消息类型又划分为私聊和群聊两种 枚举类如下：

  1. @AllArgsConstructor
  2. @Getter
  3. public enum MessageType {
  5.     //私聊
  6.     PRIVATE(1),
  7.     //群聊
  8.     GROUP(2),
  9.     ERROR(-1)
  10.     ;
  11.     private Integer type;
  13.     public static MessageType match(Integer code){
  14.         for (MessageType value : MessageType.values()) {
  15.             if (value.type.equals(code)){
  16.                 return value;
  17.             }
  18.         }
  19.         return ERROR;
  20.     }
  22. }

 

创建连接请求的拦截器

  1. import com.alibaba.fastjson2.JSON;
  2. import com.bxt.common.vo.Result;
  3. import com.bxt.demo.im.cmd.IMCommand;
  4. import com.bxt.demo.im.server.IMServer;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  8. /**
  9.  * @Description: 用户连接到服务端的拦截器
  10.  * @author: bhw
  11.  * @date: 2023年09月27日 14:28
  12.  */
  13. public class ConnectionHandler {
  14.     public static void execute(ChannelHandlerContext ctx, IMCommand command) {
  15.         if (IMServer.USERS.containsKey(command.getNickName())) {
  16.             ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error(command.getNickName() + "已经在线，不能重复连接"))));
  17.             ctx.channel().disconnect();
  18.             return;
  19.         }
  21.         IMServer.USERS.put(command.getNickName(), ctx.channel());
  23.         ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.success("系统消息：" + command.getNickName() + "与服务端连接成功"))));
  25.         ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.success(JSON.toJSONString(IMServer.USERS.keySet())))));
  26.     }
  27. }

加入群组功能的拦截器

  1. /**
  2.  * @Description: 加入群聊拦截器
  3.  * @author: bhw
  4.  * @date: 2023年09月27日 15:07
  5.  */
  6. public class JoinGroupHandler {
  7.     public static void execute(ChannelHandlerContext ctx) {
  8.         try {
  9.             IMServer.GROUP.add(ctx.channel());
  10.             ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.success("加入系统默认群组成功！"))));
  11.         } catch (Exception e) {
  12.             ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error("消息内容异常"))));
  13.         }
  15.     }
  16. }

发送聊天到指定对象的功能拦截器

  1. import com.alibaba.excel.util.StringUtils;
  2. import com.alibaba.fastjson2.JSON;
  3. import com.bxt.common.vo.Result;
  4. import com.bxt.demo.im.cmd.ChatMessage;
  5. import com.bxt.demo.im.cmd.MessageType;
  6. import com.bxt.demo.im.server.IMServer;
  7. import io.netty.channel.Channel;
  8. import io.netty.channel.ChannelHandlerContext;
  9. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  11. import java.util.Objects;
  13. /**
  14.  * @Description: 聊天拦截器
  15.  * @author: bhw
  16.  * @date: 2023年09月27日 15:07
  17.  */
  18. public class ChatHandler {
  19.     public static void execute(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
  20.         try {
  21.             ChatMessage message = JSON.parseObject(frame.text(), ChatMessage.class);
  22.             MessageType msgType = MessageType.match(message.getType());
  24.             if (msgType.equals(MessageType.PRIVATE)) {
  25.                 if (StringUtils.isBlank(message.getTarget())){
  26.                     ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error("系统消息：消息发送失败，请选择消息发送对象"))));
  27.                     return;
  28.                 }
  29.                 Channel channel = IMServer.USERS.get(message.getTarget());
  30.                 if (Objects.isNull(channel) || !channel.isActive()){
  31.                     ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error("系统消息：消息发送失败，对方不在线"))));
  32.                     IMServer.USERS.remove(message.getTarget());
  33.                     return;
  34.                 }
  35.                 channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.success("私聊消息（" + message.getTarget() + "）:" + message.getContent()))));
  37.             } else if (msgType.equals(MessageType.GROUP)) {
  38.                 IMServer.GROUP.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.success("群消息：发送者（" + message.getNickName() + "）:" + message.getContent()))));
  39.             }else {
  40.                 ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error("系统消息：不支持的消息类型"))));
  41.             }
  44.         } catch (Exception e) {
  45.             ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error("消息内容异常"))));
  46.         }
  48.     }
  49. }

最后是websocket拦截器 接收到客户端的指令后选择对应的拦截器实现相应的功能：

  1. import com.alibaba.fastjson2.JSON;
  2. import com.bxt.common.vo.Result;
  3. import com.bxt.demo.im.cmd.CommandType;
  4. import com.bxt.demo.im.cmd.IMCommand;
  5. import com.bxt.demo.im.server.IMServer;
  6. import io.netty.channel.Channel;
  7. import io.netty.channel.ChannelHandlerContext;
  8. import io.netty.channel.SimpleChannelInboundHandler;
  9. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  10. import lombok.extern.slf4j.Slf4j;
  12. /**
  13.  * @Description: websocket拦截器
  14.  * @author: bhw
  15.  * @date: 2023年09月27日 13:59
  16.  */
  17. @Slf4j
  18. public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
  20.     @Override
  21.     protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
  22.         System.out.println(frame.text());
  23.         try {
  24.             IMCommand command = JSON.parseObject(frame.text(), IMCommand.class);
  25.             CommandType cmdType = CommandType.match(command.getCode());
  26.             if (cmdType.equals(CommandType.CONNECT)){
  27.                 ConnectionHandler.execute(ctx, command);
  28.             } else if (cmdType.equals(CommandType.CHAT)) {
  29.                 ChatHandler.execute(ctx,frame);
  30.             } else if (cmdType.equals(CommandType.JOIN_GROUP)) {
  31.                 JoinGroupHandler.execute(ctx);
  32.             } else {
  33.                 ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error("不支持的code"))));
  34.             }
  35.         }catch (Exception e){
  36.             ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Result.error(e.getMessage()))));
  37.         }
  39.     }
  41.     @Override
  42.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  43.         // 当连接断开时被调用
  44.         Channel channel = ctx.channel();
  45.         // 从 USERS Map 中移除对应的 Channel
  46.         removeUser(channel);
  47.         super.channelInactive(ctx);
  48.     }
  50.     private void removeUser(Channel channel) {
  51.         // 遍历 USERS Map，找到并移除对应的 Channel
  52.         IMServer.USERS.entrySet().removeIf(entry -> entry.getValue() == channel);
  53.     }
  54. }

第三步 启动服务

  1. @SpringBootApplication
  2. public class DemoApplication {
  3.     public static void main(String[] args) {
  4.         SpringApplication.run(DemoApplication.class, args);
  5.        // 启动IM服务
  6.         try {
  7.             IMServer.start();
  8.         } catch (InterruptedException e) {
  9.             throw new RuntimeException(e);
  10.         }
  11.     }
  13. }

现在 客户端通过socket协议访问8087端口即可实现基本的聊天室功能了！

原文链接：https://www.cnblogs.com/seizedays/p/17772433.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Spring热门文章

  • 解决springboot+activemq启动报注解错误的问题
  • Springboot整合Rabbitmq之Confirm和Return机制
  • SpringBoot SpringEL表达式的使用
  • SpringBoot自动配置原理,你真的懂吗?(简单易懂)
  • Spring源码：Bean生命周期（终章）
  • Springboot-Management的项目实践
  • 解决springboot 启动找不到主类的问题
  • Spring5路径匹配器PathPattern解析
  • SpringAOP如何获取方法参数上的注解
  • java与es8实战之五：SpringBoot应用中操作es8(带安全检查：https、账号密码、API Key)

    • Spring推荐文章

  • SpringBoot Session共享，配置不生效问题排查 → 你竟然在代码里下毒！
  • 【Spring源码分析】Spring Scope功能中的动态代理 - Scoped Proxy
  • Spring AOP概念及原理
  • SpringBoot2.7还是任性的，就是不支持Logback1.3，你能奈他何
  • SpringBoot+ Sharding Sphere 轻松实现数据库字段加解密
  • Spring MVC 中视图的实现原理，在Spring MVC 中实现重定向和转发，以及访问静态资源 - Rainbow-Sea
  • Spring MVC 中 HttpMessageConverter 转换器 - Rainbow-Sea
  • SpringBoot+mail 轻松实现各类邮件自动推送
  • Spring MVC 获取三个域(request请求域，session 会话域，application 应用域)对象的方式 - Rainbow-Sea
  • SpringBoot3整合SpringDoc实现在线接口文档

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号