netty 二进制简单通信

Posted 2022-08-31 LANGZI7758521

环境：

netty版本4.0.36Final

jdk7

服务端：

1.App.java

package com.mind.core;

import com.mind.core.net.websocket.server.NettyServer;

/**
 * Hello world!
 *
 */
public class App 

    public static void main( String[] args )
    
        
        try 
			System.out.println("服务端开启等待客户端链接");
			new NettyServer().bind(8003);
		 catch (Exception e) 
			e.printStackTrace();
		
    

2.NettyServer.java

package com.mind.core.net.websocket.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioserverSocketChannel;
 
public class NettyServer 
	
 
	public void bind(int port) 
 
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workGroup = new NioEventLoopGroup();
 
		try 
 
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workGroup);
			b.channel(NioServerSocketChannel.class);
			b.childHandler(new ChildChannelHandler());

			System.out.println("服务端开启等待客户端连接 ... ...");
			// 绑定端口
			ChannelFuture f = b.bind(port).sync();
 
			// 等待服务端监听端口关闭
			f.channel().closeFuture().sync();
 
		 catch (Exception e) 
			e.printStackTrace();
		 finally 
			// 优雅的退出
			bossGroup.shutdownGracefully();
			workGroup.shutdownGracefully();
		
 
	
 

3.ChildChannelHandler.java

package com.mind.core.net.websocket.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;

public class ChildChannelHandler extends ChannelInitializer<SocketChannel> 
 
	@Override
	protected void initChannel(SocketChannel ch) throws Exception 
 
		System.out.println("报告");
		System.out.println("信息：有一客户端链接到本服务端");
		System.out.println("IP:" + ch.localAddress().getHostName());
		System.out.println("Port:" + ch.localAddress().getPort());
		System.out.println("报告完毕");
 
	
		ch.pipeline().addLast("http-codec",new HttpServerCodec());
		ch.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));
		ch.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
		ch.pipeline().addLast("handler",new MyWebSocketServerHandler());
       

4.MyWebSocketServerHander.java

package com.mind.core.net.websocket.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;

import java.util.logging.Logger;

public class MyWebSocketServerHandler extends
        SimpleChannelInboundHandler<Object>
    private static final Logger logger = Logger
            .getLogger(WebSocketServerHandshaker.class.getName());
    private WebSocketServerHandshaker handshaker;
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception 
        // 添加
        Global.group.add(ctx.channel());
        System.out.println("客户端与服务端连接开启");
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception 
        // 移除
        Global.group.remove(ctx.channel());
        System.out.println("客户端与服务端连接关闭");
    
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception 
        if (o instanceof FullHttpRequest) 
            handleHttpRequest(channelHandlerContext, ((FullHttpRequest) o));
         else if (o instanceof WebSocketFrame) 
            handlerWebSocketFrame(channelHandlerContext, (WebSocketFrame) o);
        
    
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception 
        ctx.flush();
    
    private void handlerWebSocketFrame(ChannelHandlerContext ctx,
                                       WebSocketFrame frame) 
        // 判断是否关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) 
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame
                    .retain());
        
        // 判断是否ping消息
        if (frame instanceof PingWebSocketFrame) 
            ctx.channel().write(
                    new PongWebSocketFrame(frame.content().retain()));
            return;
        
        // 测试二进制数据接收
        if(frame instanceof BinaryWebSocketFrame) 
            System.out.println("二进制数据接收");
            ByteBuf buf = frame.content();

            for (int i = 0; i < buf.capacity(); i++)
                byte b = buf.getByte(i);
                System.out.println("byte:"+b);
            

        

    
    private void handleHttpRequest(ChannelHandlerContext ctx,
                                   FullHttpRequest req) 
        if (!req.getDecoderResult().isSuccess()
                || (!"websocket".equals(req.headers().get("Upgrade")))) 
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(
                    HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws://192.168.199.89:8003/websocket", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) 
            WebSocketServerHandshakerFactory
                    .sendUnsupportedWebSocketVersionResponse(ctx.channel());
         else 
            handshaker.handshake(ctx.channel(), req);
        
    
    private static void sendHttpResponse(ChannelHandlerContext ctx,
                                         FullHttpRequest req, DefaultFullHttpResponse res) 
        // 返回应答给客户端
        if (res.getStatus().code() != 200) 
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
                    CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        
        // 如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!isKeepAlive(req) || res.getStatus().code() != 200) 
            f.addListener(ChannelFutureListener.CLOSE);
        
    
    private static boolean isKeepAlive(FullHttpRequest req) 
        return false;
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception 
        cause.printStackTrace();
        ctx.close();
    

客户端：net.js

    this._socket = new WebSocket("ws://192.168.199.89:7397/websocket");
    this._socket.binaryType.CONNECTING;
    var that = this;
    this._socket.onopen = function (event) 
        //that._socket.send("hello");
        var binary = new Uint8Array(2);
        binary[0] = 11;
        binary[1] = 12;
        that._socket.send(binary.buffer);