SpringBoot2 使用netty实现websocket

Reads: 2341 Edit

关于Netty

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

MAVEN依赖

 <!--    websocket    -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.36.Final</version>
        </dependency>
        
        
          <!--添加static和templates的依赖/html-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
        </dependency>

创建启动器

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

package com.mqtt;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import com.mqtt.demo3.NettyServer;

@SpringBootApplication
public class MqttdemoApplication {

	public static void main(String[] args) {
		SpringApplication.run(MqttdemoApplication.class, args);

		try {
			System.out.println("http://127.0.0.1:8888/ws/index");
			new NettyServer(12345).start();
		} catch (Exception e) {
			System.out.println("NettyServerError:" + e.getMessage());
		}

	}

}

NettyServer

package com.mqtt.demo3;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

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

MyChannelHandlerPool

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

package com.mqtt.demo3;

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

public class MyChannelHandlerPool  {
	 public MyChannelHandlerPool(){}

	public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

}

MyWebSocketHandler

处理ws一下几种情况:

channelActive与客户端建立连接
channelInactive与客户端断开连接
channelRead0客户端发送消息处理

MyWebSocketHandler

package com.mqtt.demo3;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.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
	    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
	        System.out.println("客户端收到服务器数据:" + msg.text());
	        sendAllMessage(msg.text());
	    }

	    private void sendAllMessage(String message){
	        //收到信息后,群发给所有channel
	        MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
	    }
}

socket.html

保存在 %/resources/static目录下

<!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>

资料参考:

https://blog.csdn.net/u013184307/article/details/95479783

关于作者

王硕,网名信平,十多年软件开发经验,业余架构师,精通Java/Python/Go等,喜欢研究技术,著有《PyQt 5 快速开发与实战》《Python 3.* 全栈开发》,多个业余开源项目托管在GitHub上,欢迎微博交流。


Comments

Make a comment

www.ultrapower.com ,王硕的博客,专注于研究互联网产品和技术,提供中文精品教程。 本网站与其它任何公司及/或商标无任何形式关联或合作。