netty实现websocket推送消息

前言

由于http协议为应答模式的连接，无法保持长连接于是引入了websocket套接字长连接概念，能够保持数据持久性的交互；本篇文章将告知读者如何使用netty实现简单的消息推送功能

websocket请求头

GET / HTTP/1.1
Host: 127.0.0.1:8096
Connection: Upgrade
Pragma: no-cache
Cache-Control: no-cache
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.82 Safari/537.36
Upgrade: websocket
Origin: http://localhost:8056
Sec-WebSocket-Version: 13

websocket请求头 会有 Connection 升级为 Upgrade, 并且Upgrade 属性值为 websocket

引入依赖

引入netty和 引擎模板依赖

    
        
            io.netty
            netty-all
            4.1.55.Final
        
        
            org.springframework.boot
            spring-boot-starter-thymeleaf

创建WebSocketServer

创建Nio线程组，并在辅助启动器中中注入 自定义处理器；定义套接字端口为8096；

/**
 * @author lsc
 * 
 

 */
@Slf4j
public class WebSocketServer {

    public void init(){
        NioEventLoopGroup boss=new NioEventLoopGroup();
        NioEventLoopGroup work=new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap=new ServerBootstrap();
            bootstrap.group(boss,work);
            bootstrap.channel(NioServerSocketChannel.class);
            // 自定义处理器
            bootstrap.childHandler(new SocketChannelInitializer());
            Channel channel = bootstrap.bind(8096).sync().channel();
            log.info("------------webSocket服务器启动成功-----------："+channel);
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.info("---------运行出错----------："+e);
        }finally {
            boss.shutdownGracefully();
            work.shutdownGracefully();
            log.info("------------websocket服务器已关闭----------------");
        }
    }
}

SocketChannelInitializer

SocketChannelInitializer 中定义了聚合器 HttpObjectAggregator 将多个http片段消息聚合成完整的http消息，并且指定大小为65536；最后注入自定义的WebSocketHandler；

/**
 * @author lsc
 * 
 

 */
public class SocketChannelInitializer extends ChannelInitializer {

    @Override
    protected void initChannel(SocketChannel ch) {
        //设置log监听器
        ch.pipeline().addLast("logging",new LoggingHandler("DEBUG"));
        //设置解码器
        ch.pipeline().addLast("http-codec",new HttpServerCodec());
        //聚合器
        ch.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));
        //用于大数据的分区传输
        ch.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
        //自定义业务handler
        ch.pipeline().addLast("handler",new WebSocketHandler());
    }
}

WebSocketHandler

WebSocketHandler 中对接收的消息进行判定，如果是websocket 消息 则将消息广播给所有通道；

/**
 * @author lsc
 * 
 

 */
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler  {

    // 存放已经连接的通道
    private  static ConcurrentMap ChannelMap=new ConcurrentHashMap();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest){

            System.out.println("------------收到http消息--------------"+msg);
            handleHttpRequest(ctx,(FullHttpRequest)msg);
        }else if (msg instanceof WebSocketFrame){
            //处理websocket客户端的消息
            String message = ((TextWebSocketFrame) msg).text();
            System.out.println("------------收到消息--------------"+message);
//            ctx.channel().writeAndFlush(new TextWebSocketFrame(message));
            // 将消息回复给所有连接
            Collection values = ChannelMap.values();
            for (Channel channel: values){
                channel.writeAndFlush(new TextWebSocketFrame(message));
            }
        }

    }

    /**
     * @author lsc
     * 
 处理http请求升级

     */
    private void handleHttpRequest(ChannelHandlerContext ctx,
                                   FullHttpRequest req) throws Exception {

        // 该请求是不是websocket upgrade请求
        if (isWebSocketUpgrade(req)) {
            String ws = "ws://127.0.0.1:8096";
            WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(ws, null, false);
            WebSocketServerHandshaker handshaker = factory.newHandshaker(req);

            if (handshaker == null) {// 请求头不合法, 导致handshaker没创建成功
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                // 响应该请求
                handshaker.handshake(ctx.channel(), req);
            }
            return;
        }
    }

    //n1.GET? 2.Upgrade头 包含websocket字符串?
    private boolean isWebSocketUpgrade(FullHttpRequest req) {
        HttpHeaders headers = req.headers();
        return req.method().equals(HttpMethod.GET)
                && headers.get(HttpHeaderNames.UPGRADE).equals("websocket");
    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //添加连接
        log.debug("客户端加入连接："+ctx.channel());
        Channel channel = ctx.channel();
        ChannelMap.put(channel.id().asShortText(),channel);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //断开连接
        log.debug("客户端断开连接："+ctx.channel());
        Channel channel = ctx.channel();
        ChannelMap.remove(channel.id().asShortText());
    }
}


最后将WebSocketServer 注入spring监听器，在服务启动的时候运行；


@Slf4j
@Component
public class ApplicationConfig implements ApplicationListener {



    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        WebSocketServer webSocketServer = new WebSocketServer();
        webSocketServer.init();
    }
}




视图转发


编写 IndexController 对视图进行转发


/**
 * @author lsc
 * 
 

 */
@Controller
public class IndexController {

    @GetMapping("index")
    public ModelAndView index(){
        ModelAndView modelAndView = new ModelAndView("socket");
        return modelAndView;
    }
}



html





    
    用户页面












附上配置文件


server:
  port: 8056

spring:
  # 引擎模板配置
  thymeleaf:
    cache: false # 关闭缓存
    mode: html # 去除htm5严格校验
    prefix: classpath:/templates/ # 指定 thymeleaf 模板路径
    encoding: UTF-8 # 指定字符集编码
    suffix: .html



运行服务后


前端页面显示消息




服务端打印消息




源码获取：知识追寻者公众号回复：netty


配套教程


    

  • netty入门(1)
    • 

  • nettyTCP 粘包拆包(2)
    • 

  • netty断线重连(3)
    • 

  • netty心跳检测机制(4)
    • 

  • netty 编码解码(5)
    • 

  • netty实现websocket推送消息(6)
    • 


						  
					  
						
							


						

							
							
			网络编程							
						



						

						
						


						

						
					
					

					
					

						
相关

						

				
						
			

																
								

										
网络编程中URL的encode和decode

									
									

								 

        

								

							
							

																
								

										
Android网络编程之Http通信

									
									

								 

        

								

							
							

																
								

										
d24_网络编程02

									
									

								 

        

								

							
							

																
								

										
网络编程

									
									

								 

        

								

							
							

																
								

										
Netty | 第1章 Java NIO 网络编程《Netty In Action》

									
									

								 

        

								

							
							

																
								

										
网络编程实战 1

									
									

								 

        

								

							
							

																
								

										
网络编程+并发编程总结

									
									

								 

        

								

							
							

																
								

										
Python网络编程—TCP套接字之HTTP传输

									
									

								 

        

								

							
							

																
								

										
Python网络编程（二）

									
									

								 

        

								

							
							

																
								

										
网络编程-HTTP cookie

									
									

								 

        

								

							
							

																
								

										
网络编程02--并发理论

									
									

								 

        

								

							
							

																
								

										
python网络编程

									
									

								 

        

								

							
							
            
            
            
                                    
           
  			
						

					

					
				

				

														
					
					

					
									

					
									   
标签

										

										
    
										

									

										
										

									

					

					
				


			
		
		
		
			


						





	

	
一品网 冀ICP备14022925号-6