SpringBoot如何集成Netty

更新時間：2024年06月17日 17:01:19 作者：Sea-Man

這篇文章主要介紹了SpringBoot如何集成Netty問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

一、pom依賴

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.77.Final</version>
</dependency>
<dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-all</artifactId>
    <version>5.5.8</version>
</dependency>

二、配置yml文件

server:
  port: 8001
  servlet:
    context-path: /netty
netty:
  url: 0.0.0.0  #0.0.0.0表示綁定任意ip
  port: 20004

三、服務(wù)端

package com.tlxy.lhn.controller.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        //創(chuàng)建兩個線程組bossGroup和workerGroup，含有的子線程NioEventLoop的個數(shù)默認是CPU的兩倍
        //bossGroup只是處理連接請求，真正的和客戶端業(yè)務(wù)處理，會交給workerGroup完成
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(1);

        try {
            //創(chuàng)建服務(wù)器端的啟動對象
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用鏈式編程來配置參數(shù)
            bootstrap.group(bossGroup, workerGroup)//設(shè)置兩個線程組
                    .channel(NioServerSocketChannel.class)//使用NioServerSocketChannel作為服務(wù)器的通道實現(xiàn)
                    //初始化服務(wù)器連接隊列大小，服務(wù)端處理客戶端連接請求是順序處理的，所以同一時間只能處理一個客戶端連接
                    //多個客戶端同時來的時候，服務(wù)端將不能處理的客戶端連接請求放在隊列中等待處理
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            //對workerGroup的SocketChannel設(shè)置處理器
                            channel.pipeline().addLast(new NettyServerHandler());
                        }
                    });

            System.out.println("netty server start..");

            //綁定一個端口并且同步生成一個ChannelFuture異步對象，通過isDone()等方法可以判斷異步事件的執(zhí)行情況
            //啟動服務(wù)器（并綁定的端口），bind是異步操作，sync方法是等待異步操作執(zhí)行完畢
            ChannelFuture cf = bootstrap.bind(9000).sync();

            //給cf注冊監(jiān)聽器，監(jiān)聽我們關(guān)心的事件
            cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (cf.isSuccess()) {
                        System.out.println("監(jiān)聽端口9000成功");
                    } else {
                        System.out.println("監(jiān)聽端口9000失敗");
                    }
                }
            });
            //等待服務(wù)端監(jiān)聽端口關(guān)閉，closeFuture是異步操作
            //通過sync方法同步等待通道關(guān)閉處理完畢，這里會阻塞等待通道關(guān)閉完成，內(nèi)部調(diào)用的是Object的wait()方法
            cf.channel().closeFuture().sync();

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }


    }
}

NettyServer類中的

channel.pipeline().addLast(new NettyServerHandler());

對應(yīng)以下的處理器。

package com.tlxy.lhn.controller.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客戶端發(fā)送消息是：" + buf.toString(CharsetUtil.UTF_8));
        // 讀取byteBuf
        // 業(yè)務(wù)處理
        // 回消息給客戶端
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ByteBuf buf = Unpooled.copiedBuffer("HelloClient".getBytes(CharsetUtil.UTF_8));
        ctx.writeAndFlush(buf);
    }

    //只要Netty拋出錯誤就會執(zhí)行,Netty斷會開連接會拋出連接超時的錯誤
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.info("關(guān)閉通道");
        cause.printStackTrace();
        ctx.close();
    }
}

四、客戶端

package com.tlxy.lhn.controller.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        //客戶端需要一個事件循環(huán)組
        NioEventLoopGroup group = new NioEventLoopGroup();

        try {
            //創(chuàng)建客戶端啟動對象
            //注意客戶端使用的不是SocketBootstrap而是Bootstrap
            Bootstrap bootstrap = new Bootstrap();

            // 設(shè)置相關(guān)參數(shù)
            bootstrap.group(group) //設(shè)置線程組
                    .channel(NioSocketChannel.class)// 使用NioSocketChannel作為客戶端的通道實現(xiàn)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyClientHandler());
                        }
                    });

            System.out.println("netty client start..");
            ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
            cf.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully();
        }


    }
}

NettyClient類中

ch.pipeline().addLast(new NettyClientHandler());

為處理器。

package com.tlxy.lhn.controller.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 客戶端連接標識
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
        ctx.writeAndFlush(buf);
    }

    //當通道建立后有事件時會觸發(fā)，即服務(wù)端發(fā)送數(shù)據(jù)給客戶端
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("收到服務(wù)端的消息是：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服務(wù)端地址是：" + ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("關(guān)閉通道");
        cause.printStackTrace();
        ctx.close();
    }

}

五、粘包和拆包問題

客戶端和服務(wù)端都是固定的框架，我們只需寫處理器。

粘包和拆包問題，可以自己手寫通過固定長度發(fā)送數(shù)據(jù)，或者使用Google的Protostuff。

<dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-api</artifactId>
    <version>1.0.8</version>
</dependency>
<dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-core</artifactId>
    <version>1.0.8</version>
</dependency>
<dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-runtime</artifactId>
    <version>1.0.8</version>
</dependency>

package com.tlxy.lhn.controller.netty;

import com.dyuproject.protostuff.LinkedBuffer;
import com.dyuproject.protostuff.ProtostuffIOUtil;
import com.dyuproject.protostuff.Schema;
import com.dyuproject.protostuff.runtime.RuntimeSchema;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ProtostuffUtil {
    private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<Class<?>, Schema<?>>();

    private static <T> Schema<T> getSchema(Class<T> clazz) {
        @SuppressWarnings("unchecked")
        Schema<T> schema = (Schema<T>) cachedSchema.get(clazz);
        if (schema == null) {
            schema = RuntimeSchema.getSchema(clazz);
            if (schema != null) {
                cachedSchema.put(clazz, schema);
            }
        }
        return schema;
    }

    /**
     * 序列化
     *
     * @param obj
     * @return
     */
    public static <T> byte[] serializer(T obj) {
        @SuppressWarnings("unchecked")
        Class<T> clazz = (Class<T>) obj.getClass();
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        try {
            Schema<T> schema = getSchema(clazz);
            return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        } finally {
            buffer.clear();
        }
    }

    /**
     * 反序列化
     *
     * @param data
     * @param clazz
     * @return
     */
    public static <T> T deserializer(byte[] data, Class<T> clazz) {
        try {
            T obj = clazz.newInstance();
            Schema<T> schema = getSchema(clazz);
            ProtostuffIOUtil.mergeFrom(data, obj, schema);
            return obj;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

}