SpringBoot+Netty+WebSocket 实现消息推送

简介: SpringBoot+Netty+WebSocket 实现消息推送

关于Netty

Netty 是一个利用 Java 的高级网络的能力,隐藏底层的复杂性而提供一个易于使用/使用其服务器的 API 的客户端框架。

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

Maven依赖

<dependencies>
 <!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
 <dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-all</artifactId>
  <version>4.1.36.Final</version>
 </dependency>
</dependencies>

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

SpringBoot应用程序

启动器中需要new一个NettyServer,并显式调用netty。

@SpringBootApplication
public class SpringCloudStudyDemoApplication {
 public static void main(String[] args) {
  SpringApplication.run(SpringCloudStudyDemoApplication.class,args);
  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());
  }
 }
}

网络服务器

启动的NettyServer,这里进行配置

/**
 * NettyServer Netty服务器配置
 * @author zhengkai.blog.csdn.net
 * @date 2019-06-12
 */
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>() { // 绑定客户端连接时候触发操作
                        @Override
                        protected 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 MyWebSocketHandler());
                        }
                    });
            ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
            System.out.println(NettyServer.class + " 启动正在监听: " + cf.channel().localAddress());
            cf.channel().closeFuture().sync(); // 关闭服务器通道
        } finally {
            group.shutdownGracefully().sync(); // 释放线程池资源
            bossGroup.shutdownGracefully().sync();
        }
    }
}

MyChannelHandlerPool

通道组池,管理所有websocket连接

/**
 * MyChannelHandlerPool
 * 通道组池,管理所有websocket连接
 * @author zhengkai.blog.csdn.net
 * @date 2019-06-12
 */
public class MyChannelHandlerPool {
    public MyChannelHandlerPool(){}
    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}

MyWebSocketHandler

处理ws一下几种情况:

  • channelActive与客户端建立连接
  • channelInactive与客户端连接破坏
  • channelRead0客户端发送消息处理
/**
 * NettyServer Netty服务器配置
 * @author zhengkai.blog.csdn.net
 * @date 2019-06-12
 */
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>() { // 绑定客户端连接时候触发操作
                        @Override
                        protected 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", "WebSocket", true, 65536 * 10));
                            ch.pipeline().addLast(new MyWebSocketHandler());
                        }
                    });
            ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
            System.out.println(NettyServer.class + " 启动正在监听: " + cf.channel().localAddress());
            cf.channel().closeFuture().sync(); // 关闭服务器通道
        } finally {
            group.shutdownGracefully().sync(); // 释放线程池资源
            bossGroup.shutdownGracefully().sync();
        }
    }
}

套接字.html

主要是连接ws,发送消息,以及消息反馈

<!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.net
        var 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>

控制器

写好当然html还需要一个控制器来引导页面。

@RestController
public class IndexController {
 @GetMapping("/index")
 public ModelAndView  index(){
  ModelAndView mav=new ModelAndView("socket");
  mav.addObject("uid", RandomUtil.randomNumbers(6));
  return mav;
 }
}

效果展示

微信图片_20220907172256.png

微信图片_20220907172259.png

微信图片_20220907172301.png

netty 改造支持url参数

1.首先,调整一下加载处理程序的顺序,优先MyWebSocketHandlerWebSocketServerProtocolHandler上面。

ch.pipeline().addLast(new MyWebSocketHandler());
ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));

2.之后,会通过某种方式,会通过某种方式,获取一个完整的URL地址,是一个接受信息的MyWebSocketHandler话。channelRead``FullHttpRequest``FullHttpRequest.uri()``TextWebSocketFrame

public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端建立连接,通道开启!");
        //添加到channelGroup通道组
        MyChannelHandlerPool.channelGroup.add(ctx.channel());
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端断开连接,通道关闭!");
        //添加到channelGroup 通道组
        MyChannelHandlerPool.channelGroup.remove(ctx.channel());
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //首次连接是FullHttpRequest,处理参数 by zhengkai.blog.csdn.net
        if (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);
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
    }
    private void sendAllMessage(String message){
        //收到信息后,群发给所有channel
        MyChannelHandlerPool.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;
        }
    }
}

3.html中的ws地址也进行了改造

socket = new WebSocket("ws://127.0.0.1:12345/ws?uid=666&gid=777");
  1. 改造后输出情况
收到新连接
与客户端建立连接,通道开启!
接收到的参数是:{"uid":"666","gid":"777"}
/ws
客户端收到服务器数据:142531:这里输入消息
客户端收到服务器数据:142531:这里输入消息
客户端收到服务器数据:142531:这里输入消息

失败:WebSocket 打开握手超时

听说是 ssl wss 的情况下会出现,来自@around-gao 的解决方法:

MyWebSocketHandlerWebSocketServerProtocolHandler调下顺序就好了。

相关文章
|
2月前
|
人工智能 前端开发 JavaScript
SpringBoot实现网页消息推送的5种方法
本文详细介绍了在SpringBoot中实现网页消息推送的几种主流方案,包括短轮询、长轮询、SSE(Server-Sent Events)、WebSocket以及STOMP。每种方案各有优缺点,适用于不同的场景需求。短轮询简单易实现但效率低;长轮询提升了实时性但仍有限制;SSE适合单向通信且轻量高效;WebSocket支持全双工通信,适合高实时性要求的场景;STOMP基于WebSocket,提供更高级的消息传递功能。通过对比分析,开发者可根据业务需求、性能要求及浏览器兼容性选择最适合的技术方案,同时可结合多种技术实现优雅降级,优化用户体验。
305 57
|
1月前
|
网络协议 Java
在SpringBoot项目中使用Netty实现远程调用
本文介绍了使用Netty解决网络连接性能问题的方法,重点讲解了Netty的NIO特性及其在SpringBoot中的应用。Netty作为高效的NIO框架,支持非阻塞IO,能通过单线程管理多个客户端连接,简化TCP/UDP套接字服务器开发。文章详细展示了Netty在SpringBoot中实现远程调用的过程,包括服务端与客户端代码实现、依赖配置及测试验证。通过示例代码,如`NettyServer`、`NettyClientUtil`等,清晰说明了Netty的工作原理和实际应用,解决了半包等问题,并提供了完整的测试结果。
151 3
|
2月前
|
前端开发 网络协议 应用服务中间件
Netty基础—7.Netty实现消息推送服务
本文详细介绍了如何使用Netty实现HTTP服务器、WebSocket以及基于WebSocket的消息推送系统。首先,通过解析HTTP请求和响应消息,展示了Netty在性能和可靠性上的优势,并提供了具体代码示例。接着,分析了HTTP协议的弊端及Ajax短轮询的不足,引出WebSocket全双工通信的优势,包括连接建立、数据处理逻辑与ping-pong探测等。最后,构建了一个完整的消息推送系统,涵盖PushServer、运营客户端与浏览器客户端的交互过程,实现了全连接推送和实时消息传递。
|
9月前
|
开发框架 前端开发 网络协议
Spring Boot结合Netty和WebSocket,实现后台向前端实时推送信息
【10月更文挑战第18天】 在现代互联网应用中,实时通信变得越来越重要。WebSocket作为一种在单个TCP连接上进行全双工通信的协议,为客户端和服务器之间的实时数据传输提供了一种高效的解决方案。Netty作为一个高性能、事件驱动的NIO框架,它基于Java NIO实现了异步和事件驱动的网络应用程序。Spring Boot是一个基于Spring框架的微服务开发框架,它提供了许多开箱即用的功能和简化配置的机制。本文将详细介绍如何使用Spring Boot集成Netty和WebSocket,实现后台向前端推送信息的功能。
2019 1
|
5月前
|
消息中间件 XML 前端开发
springBoot集成websocket实时消息推送
本文介绍了如何在Spring Boot项目中集成WebSocket实现实时消息推送。首先,通过引入`spring-boot-starter-websocket`依赖,配置`WebSocketConfig`类来启用WebSocket支持。接着,创建`WebSocketTest`服务器类,处理连接、消息收发及错误等事件,并使用`ConcurrentHashMap`管理用户连接。最后,前端通过JavaScript建立WebSocket连接,监听消息并进行相应处理。此方案适用于需要实时通信的应用场景,如聊天室、通知系统等。
296 2
|
9月前
|
NoSQL Java Redis
redis的基本命令,并用netty操作redis(不使用springboot或者spring框架)就单纯的用netty搞。
这篇文章介绍了Redis的基本命令,并展示了如何使用Netty框架直接与Redis服务器进行通信,包括设置Netty客户端、编写处理程序以及初始化Channel的完整示例代码。
204 1
redis的基本命令,并用netty操作redis(不使用springboot或者spring框架)就单纯的用netty搞。
|
9月前
|
负载均衡 网络协议 C#
C#实现WebSocket实时消息推送技术详解
C#实现WebSocket实时消息推送技术详解
513 1
|
11月前
|
人工智能 数据库连接 Go
Golang 搭建 WebSocket 应用(五) - 消息推送日志
Golang 搭建 WebSocket 应用(五) - 消息推送日志
91 1
|
11月前
|
人工智能 JSON 安全
Golang 搭建 WebSocket 应用(三) - 实现一个消息推送中心
Golang 搭建 WebSocket 应用(三) - 实现一个消息推送中心
183 0
|
2月前
|
JavaScript 前端开发 Java
制造业ERP源码,工厂ERP管理系统,前端框架:Vue,后端框架:SpringBoot
这是一套基于SpringBoot+Vue技术栈开发的ERP企业管理系统,采用Java语言与vscode工具。系统涵盖采购/销售、出入库、生产、品质管理等功能,整合客户与供应商数据,支持在线协同和业务全流程管控。同时提供主数据管理、权限控制、工作流审批、报表自定义及打印、在线报表开发和自定义表单功能,助力企业实现高效自动化管理,并通过UniAPP实现移动端支持,满足多场景应用需求。
245 1