Netty 網絡編程的秘籍,看這一篇夠了 環球速訊

2023-06-02 11:26:48    來源:Java技術指北    
Netty

version: 4.1.55.Final


(相關資料圖)

傳統的IO模型的web容器,比如老版本的Tomcat,為了增加系統的吞吐量,需要不斷增加系統核心線程數量,或者通過水平擴展服務器數量,來增加系統處理請求的能力。 有了NIO之后,一個線程即可處理多個連接事件,基于多路復用模型的Netty框架,不僅降低了使用NIO的復雜度,

優點

Netty是一款以java NIO為基礎,基于事件驅動模型支持異步、高并發的網絡應用框架

API使用簡單,開發門檻低,簡化了NIO開發網絡程序的復雜度功能強大,預置多種編解碼功能,支持多種主流協議,比如Http、WebSocket。定制能力強,可以通過ChannelHandler對通信框架靈活擴展。性能高,支持異步非阻塞通信模型成熟穩定,社區活躍,已經修復了Java NIO所有的Bug。經歷了大規模商業應用的考驗,質量有保證。IO模型

select、poll和epoll

操作系統內核基于這些函數實現非阻塞IO,以此實現多路復用模型

select

select

select 調用需要傳入 fd 數組,需要拷貝一份到內核,高并發場景下這樣的拷貝消耗的資源是驚人的。(可優化為不復制)select 在內核層仍然是通過遍歷的方式檢查文件描述符的就緒狀態,是個同步過程,只不過無系統調用切換上下文的開銷。(內核層可優化為異步事件通知)select 僅僅返回可讀文件描述符的個數,具體哪個可讀還是要用戶自己遍歷。(可優化為只返回給用戶就緒的文件描述符,無需用戶做無效的遍歷)pool

和 select 的主要區別就是,去掉了 select 只能監聽 1024 個文件描述符的限制

epool

epool

內核中保存一份文件描述符集合,無需用戶每次都重新傳入,只需告訴內核修改的部分即可。內核不再通過輪詢的方式找到就緒的文件描述符,而是通過異步 IO 事件喚醒。內核僅會將有 IO 事件的文件描述符返回給用戶,用戶也無需遍歷整個文件描述符集合。Reactor模型一、單Reactor單線程

1)可以實現通過一個阻塞對象監聽多個鏈接請求

2)Reactor對象通過select監聽客戶端請求事件,通過dispatch進行分發

3)如果是建立鏈接請求,則由Acceptor通過accept處理鏈接請求,然后創建一個Handler對象處理完成鏈接后的各種事件

4)如果不是鏈接請求,則由Reactor分發調用鏈接對應的Handler來處理

5)Handler會完成Read->業務處理->send的完整業務流程

二、單Reactor多線程

1)Reactor對象通過select監聽客戶端請求事件,收到事件后,通過dispatch分發

2)如果是建立鏈接請求,則由Acceptor通過accept處理鏈接請求,然后創建一個Handler對象處理完成鏈接后的各種事件

3)如果不是鏈接請求,則由Reactor分發調用鏈接對應的Handler來處理

4)Handler只負責事件響應不做具體業務處理

5)通過read讀取數據后,分發到worker線程池處理,處理完成后返回給Handler,Handler收到后,通過send將結果返回給client

三、主從Reactor多線程

1)Reactor主線程MainReactor對象通過select監聽鏈接事件,通過Acceptor處理

2)當Acceptor處理鏈接事件后,MainReactor將鏈接分配給SubReactor

3)SubReactor將鏈接加入到隊列進行監聽,并創建Handler進行事件處理

4)當有新事件發生時,SubReactor就會調用對應的Handler處理

5)Handler通過read讀取數據,分發到worker線程池處理,處理完成后返回給Handler,Handler收到后,通過send將結果返回給client

6)Reactor主線程可以對應多個Reactor子線程

三種模式用生活案例來理解

1)單Reactor單線程,前臺接待員和服務員是同一個人,全程為顧客服務

2)單Reactor多線程,1個前臺接待員,多個服務員,接待員只負責接待

3)主從Reactor多線程,多個前臺接待員,多個服務員

Reactor模型具有如下優點

1)響應快,不必為單個同步事件所阻塞,雖然Reactor本身依然是同步的

2)可以最大程度的避免復雜的多線程及同步問題,并且避免了多線程/進程的切換開銷

3)擴展性好,可以方便的通過增加Reactor實例個數來充分利用CPU資源

4)復用性好,Reactor模型本身與具體事件處理邏輯無關,具有很高的復用性

核心組件

1.Bootstrap 一個Netty應用通常由一個Bootstrap開始,它主要作用是配置整個Netty程序,串聯起各個組件。

Handler,為了支持各種協議和處理數據的方式,便誕生了Handler組件。Handler主要用來處理各種事件,這里的事件很廣泛,比如可以是連接、數據接收、異常、數據轉換等。

2.ChannelInboundHandler 一個最常用的Handler。這個Handler的作用就是處理接收到數據時的事件,也就是說,我們的業務邏輯一般就是寫在這個Handler里面的,ChannelInboundHandler就是用來處理我們的核心業務邏輯。

3.ChannelInitializer 當一個鏈接建立時,我們需要知道怎么來接收或者發送數據,當然,我們有各種各樣的Handler實現來處理它,那么ChannelInitializer便是用來配置這些Handler,它會提供一個ChannelPipeline,并把Handler加入到ChannelPipeline。

4.ChannelPipeline 一個Netty應用基于ChannelPipeline機制,這種機制需要依賴于EventLoop和EventLoopGroup,因為它們三個都和事件或者事件處理相關。

EventLoops的目的是為Channel處理IO操作,一個EventLoop可以為多個Channel服務。

EventLoopGroup會包含多個EventLoop。

5.Channel 代表了一個Socket鏈接,或者其它和IO操作相關的組件,它和EventLoop一起用來參與IO處理。

6.Future 在Netty中所有的IO操作都是異步的,因此,你不能立刻得知消息是否被正確處理,但是我們可以過一會等它執行完成或者直接注冊一個監聽,具體的實現就是通過Future和ChannelFutures,他們可以注冊一個監聽,當操作執行成功或失敗時監聽會自動觸發。

示例

通過一個簡單的示例,首先了解怎么基于netty開發一個通信程序,包括服務的與客戶端:

Server:

@Slf4jpublic class Server {    private EventLoopGroup boosGroup;    private EventLoopGroup workGroup;    public Server(int port){        try {            init(port);            log.info("----- 服務啟動成功 -----");        } catch (InterruptedException e) {            log.error("啟動服務出錯:{}", e.getCause());        }    }    private void init(int port) throws InterruptedException {        // 處理連接        this.boosGroup = new NioEventLoopGroup();        // 處理業務        this.workGroup = new NioEventLoopGroup();        ServerBootstrap bootstrap = new ServerBootstrap();        // 綁定        bootstrap.group(boosGroup, workGroup)                .channel(NioServerSocketChannel.class) //配置服務端                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)                .option(ChannelOption.SO_BACKLOG, 1024)                .childOption(ChannelOption.TCP_NODELAY, true)                .childOption(ChannelOption.SO_RCVBUF, 1024)                .childOption(ChannelOption.SO_SNDBUF, 1024)                .childHandler(new ChannelInitializer() {                    @Override                    protected void initChannel(SocketChannel socketChannel) throws Exception {                        socketChannel.pipeline().addLast(new ServerHandler());                    }                });        ChannelFuture channelFuture = bootstrap.bind(port).sync();        channelFuture.channel().closeFuture().sync();    }    public void close(){        this.boosGroup.shutdownGracefully();        this.workGroup.shutdownGracefully();    }}@Slf4jclass ServerHandler extends ChannelInboundHandlerAdapter {    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        log.info(">>>>>>> server active");    }    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {        //1. 讀取客戶端的數據(緩存中去取并打印到控制臺)        ByteBuf buf = (ByteBuf) msg;        byte[] request = new byte[buf.readableBytes()];        buf.readBytes(request);        String requestBody = new String(request, "utf-8");        log.info(">>>>>>>>> receive message: {}", requestBody);        //2. 返回響應數據        ctx.writeAndFlush(Unpooled.copiedBuffer((requestBody+" too").getBytes()));    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        super.exceptionCaught(ctx, cause);    }}

Client:

@Slf4jpublic class Client {    private EventLoopGroup workGroup;    private ChannelFuture channelFuture;    public Client(int port){        init(port);    }    private void init(int port){        this.workGroup = new NioEventLoopGroup();        Bootstrap bootstrap = new Bootstrap();        bootstrap.group(workGroup)                .channel(NioSocketChannel.class)                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)                .option(ChannelOption.SO_RCVBUF, 1024)                .option(ChannelOption.SO_SNDBUF, 1024)                .handler(new ChannelInitializer() {                    @Override                    protected void initChannel(SocketChannel socketChannel) throws Exception {                        socketChannel.pipeline().addLast(new ClientHandler());                    }                });        this.channelFuture = bootstrap.connect("127.0.0.1", port).syncUninterruptibly();    }    /**     *     * @param message     */    public void send(String message){        this.channelFuture.channel().writeAndFlush(Unpooled.copiedBuffer(message.getBytes()));    }    /**     *     */    public void close(){        try {            channelFuture.channel().closeFuture().sync();        } catch (InterruptedException e) {            throw new RuntimeException(e);        }        workGroup.shutdownGracefully();    }}@Slf4jclass ClientHandler extends ChannelInboundHandlerAdapter {    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        log.info(">>>>>>> client active");    }    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {        try {            ByteBuf buf = (ByteBuf) msg;            byte[] req = new byte[buf.readableBytes()];            buf.readBytes(req);            String body = new String(req, "utf-8");            log.info(">>>>>>>>> receive message: {}", body);        } finally {            ReferenceCountUtil.release(msg);        }    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        super.exceptionCaught(ctx, cause);    }}

測試:

public class StarterTests {    static int port = 9011;    @Test    public void startServer(){        Server server = new Server(9011);    }    @Test    public void startClient(){        Client client = new Client(port);        client.send("Hello Netty!");        while (true){}    }}
生態DubboSpring Reactive類似技術

Mina、Netty、Grizzly

其他

Proactor非阻塞異步網絡模型

參考

https://mp.weixin.qq.com/s?__biz=MzUxNDA1NDI3OA==&mid=2247492766&idx=2&sn=b5df49147561e467fa5677b5bb09dacb&chksm=f9496577ce3eec61383994499d96a7f2b091b5eb8ee1ac47ad021f78072ae710f41d38257406&scene=27

https://blog.csdn.net/a745233700/article/details/122660246

關鍵詞:
[責任編輯:h001]

資訊播報

聯系我們:435 226 40 @qq.com

版權所有 重播新聞網 www.porb.cn 京ICP備2022022245號-17