SpringBoot+Netty+WebSocket 实现消息推送

关于Netty

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

Maven依赖

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

SpringBootApplication

启动器中需要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,这里进行配置

/** 
* 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();
        }
    }}

socket.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>

Controller

写好了html当然还需要一个controller来引导页面。

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

效果演示

图片.png
图片.png
图片.png

改造netty支持url参数

1.首先,调整一下加载handler的顺序,优先MyWebSocketHandlerWebSocketServerProtocolHandler之上。扩展:接私活儿

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

2.其次,改造MyWebSocketHandlerchannelRead方法,首次连接会是一个FullHttpRequest类型,可以通过FullHttpRequest.uri()获取完整ws的URL地址,之后接受信息的话,会是一个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");

4.改造后控制台输出情况

收到新连接与客户端建立连接,通道开启!接收到的参数是:{"uid":"666","gid":"777"}/ws客户端收到服务器数据:142531:这里输入消息客户端收到服务器数据:142531:这里输入消息客户端收到服务器数据:142531:这里输入消息

failed: WebSocket opening handshake timed out

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

MyWebSocketHandlerWebSocketServerProtocolHandler调下顺序就好了。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,723评论 6 481
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,485评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 152,998评论 0 344
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,323评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,355评论 5 374
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,079评论 1 285
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,389评论 3 400
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,019评论 0 259
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,519评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,971评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,100评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,738评论 4 324
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,293评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,289评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,517评论 1 262
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,547评论 2 354
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,834评论 2 345

推荐阅读更多精彩内容