SpringBoot+Netty+WebSocket 实现消息推送

2022-07-26 17:47:58 浏览数 (1)

点击上方“芋道源码”，选择“设为星标”

管她前浪，还是后浪？

能浪的浪，才是好浪！

每天 10:33 更新文章，每天掉亿点点头发...

源码精品专栏

  • 原创 | Java 2021 超神之路，很肝~
  • 中文详细注释的开源项目
  • RPC 框架 Dubbo 源码解析
  • 网络应用框架 Netty 源码解析
  • 消息中间件 RocketMQ 源码解析
  • 数据库中间件 Sharding-JDBC 和 MyCAT 源码解析
  • 作业调度中间件 Elastic-Job 源码解析
  • 分布式事务中间件 TCC-Transaction 源码解析
  • Eureka 和 Hystrix 源码解析
  • Java 并发源码

来源：blog.csdn.net/qq_16504067/

article/details/119108685

  • 关于Netty
  • Maven依赖
  • SpringBoot应用程序
  • 网络服务器
  • MyChannelHandlerPool
  • MyWebSocketHandler
  • 套接字.html
  • 控制器
  • 效果展示
  • netty 改造支持url参数

关于Netty

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

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

  • 项目地址：https://gitee.com/zhijiantianya/ruoyi-vue-pro
  • 视频教程：https://doc.iocoder.cn/video/

Maven依赖

代码语言：javascript复制
<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 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

  • 项目地址：https://gitee.com/zhijiantianya/yudao-cloud
  • 视频教程：https://doc.iocoder.cn/video/

SpringBoot应用程序

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

代码语言：javascript复制
@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，这里进行配置

代码语言：javascript复制
/**
 * 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连接

代码语言：javascript复制
/**
 * 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客户端发送消息处理
代码语言：javascript复制
/**
 * 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，发送消息，以及消息反馈

代码语言：javascript复制
<!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 "rn";
            };
            socket.onopen = function(event){
                var ta = document.getElementById('responseText');
                ta.value = "Netty-WebSocket服务器。。。。。。连接  rn";
            };
            socket.onclose = function(event){
                var ta = document.getElementById('responseText');
                ta.value = "Netty-WebSocket服务器。。。。。。关闭 rn";
            };
        }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还需要一个控制器来引导页面。

代码语言：javascript复制
@RestController
public class IndexController {
 
 @GetMapping("/index")
 public ModelAndView  index(){
  ModelAndView mav=new ModelAndView("socket");
  mav.addObject("uid", RandomUtil.randomNumbers(6));
  return mav;
 }
 
}

效果展示

netty 改造支持url参数

1.首先，调整一下加载处理程序的顺序，优先MyWebSocketHandler在WebSocketServerProtocolHandler上面。

代码语言：javascript复制
ch.pipeline().addLast(new MyWebSocketHandler());
ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));

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

代码语言：javascript复制
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地址也进行了改造

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

失败：WebSocket 打开握手超时

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

把MyWebSocketHandler和WebSocketServerProtocolHandler调下顺序就好了。

欢迎加入我的知识星球，一起探讨架构，交流源码。加入方式，长按下方二维码噢：

已在知识星球更新源码解析如下：

最近更新《芋道 SpringBoot 2.X 入门》系列，已经 101 余篇，覆盖了 MyBatis、Redis、MongoDB、ES、分库分表、读写分离、SpringMVC、Webflux、权限、WebSocket、Dubbo、RabbitMQ、RocketMQ、Kafka、性能测试等等内容。

提供近 3W 行代码的 SpringBoot 示例，以及超 4W 行代码的电商微服务项目。

获取方式：点“在看”，关注公众号并回复 666 领取，更多内容陆续奉上。

代码语言：javascript复制
文章有帮助的话，在看，转发吧。谢谢支持哟 (*^__^*）
网站 socket编程 android spring mybatis

0 人点赞