SpringBoot項(xiàng)目中使用Netty實(shí)現(xiàn)遠(yuǎn)程調(diào)用

來(lái)源：https://mp.weixin.qq.com/s/gNj-XQuo8vLy3bymcv-tGA

前言

眾所周知我們?cè)谶M(jìn)行網(wǎng)絡(luò)連接的時(shí)候，建立套接字連接是一個(gè)非常消耗性能的事情，特別是在分布式的情況下，用線程池去保持多個(gè)客戶(hù)端連接，是一種非常消耗線程的行為。

那么我們?cè)撏ㄟ^(guò)什么技術(shù)去解決上述的問(wèn)題呢，那么就不得不提一個(gè)網(wǎng)絡(luò)連接的利器——Netty.

正文

Netty

Netty是一個(gè)NIO客戶(hù)端服務(wù)器框架：

• 它可快速輕松地開(kāi)發(fā)網(wǎng)絡(luò)應(yīng)用程序，例如協(xié)議服務(wù)器和客戶(hù)端。
• 它極大地簡(jiǎn)化和簡(jiǎn)化了網(wǎng)絡(luò)編程，例如TCP和UDP套接字服務(wù)器。

NIO是一種非阻塞IO ，它具有以下的特點(diǎn)

• 單線程可以連接多個(gè)客戶(hù)端。
• 選擇器可以實(shí)現(xiàn)但線程管理多個(gè)Channel，新建的通道都要向選擇器注冊(cè)。
• 一個(gè)SelectionKey鍵表示了一個(gè)特定的通道對(duì)象和一個(gè)特定的選擇器對(duì)象之間的注冊(cè)關(guān)系。
• selector進(jìn)行select()操作可能會(huì)產(chǎn)生阻塞，但是可以設(shè)置阻塞時(shí)間，并且可以用wakeup()喚醒selector，所以NIO是非阻塞IO。

Netty模型selector模式

它相對(duì)普通NIO的在性能上有了提升，采用了：

• NIO采用多線程的方式可以同時(shí)使用多個(gè)selector
• 通過(guò)綁定多個(gè)端口的方式，使得一個(gè)selector可以同時(shí)注冊(cè)多個(gè)ServerSocketServer
• 單個(gè)線程下只能有一個(gè)selector，用來(lái)實(shí)現(xiàn)Channel的匹配及復(fù)用

半包問(wèn)題

TCP/IP在發(fā)送消息的時(shí)候，可能會(huì)拆包，這就導(dǎo)致接收端無(wú)法知道什么時(shí)候收到的數(shù)據(jù)是一個(gè)完整的數(shù)據(jù)。在傳統(tǒng)的BIO中在讀取不到數(shù)據(jù)時(shí)會(huì)發(fā)生阻塞，但是NIO不會(huì)。

為了解決NIO的半包問(wèn)題，Netty在Selector模型的基礎(chǔ)上，提出了reactor模式，從而解決客戶(hù)端請(qǐng)求在服務(wù)端不完整的問(wèn)題。

netty模型reactor模式

在selector的基礎(chǔ)上解決了半包問(wèn)題。

上圖，簡(jiǎn)單地可以描述為"boss接活，讓work干"：manReactor用來(lái)接收請(qǐng)求（會(huì)與客戶(hù)端進(jìn)行握手驗(yàn)證），而subReactor用來(lái)處理請(qǐng)求（不與客戶(hù)端直接連接）。另外，歡迎關(guān)注公眾號(hào)Java筆記蝦，后臺(tái)回復(fù)“后端面試”，送你一份面試題寶典！

SpringBoot使用Netty實(shí)現(xiàn)遠(yuǎn)程調(diào)用

maven依賴(lài)

<dependency>
??<groupId>org.projectlombokgroupId>
??<artifactId>lombokartifactId>
??<version>1.18.2version>
??<optional>trueoptional>
dependency>


<dependency>
??<groupId>io.nettygroupId>
??<artifactId>netty-allartifactId>
??<version>4.1.17.Finalversion>
dependency>

服務(wù)端部分

NettyServer.java：服務(wù)啟動(dòng)監(jiān)聽(tīng)器

@Slf4j
public?class?NettyServer?{
????public?void?start()?{
????????InetSocketAddress?socketAddress?=?new?InetSocketAddress("127.0.0.1",?8082);
????????//new?一個(gè)主線程組
????????EventLoopGroup?bossGroup?=?new?NioEventLoopGroup(1);
????????//new?一個(gè)工作線程組
????????EventLoopGroup?workGroup?=?new?NioEventLoopGroup(200);
????????ServerBootstrap?bootstrap?=?new?ServerBootstrap()
????????????????.group(bossGroup,?workGroup)
????????????????.channel(NioServerSocketChannel.class)
????????????????.childHandler(new?ServerChannelInitializer())
????????????????.localAddress(socketAddress)
????????????????//設(shè)置隊(duì)列大小
????????????????.option(ChannelOption.SO_BACKLOG,?1024)
????????????????//?兩小時(shí)內(nèi)沒(méi)有數(shù)據(jù)的通信時(shí),TCP會(huì)自動(dòng)發(fā)送一個(gè)活動(dòng)探測(cè)數(shù)據(jù)報(bào)文
????????????????.childOption(ChannelOption.SO_KEEPALIVE,?true);
????????//綁定端口,開(kāi)始接收進(jìn)來(lái)的連接
????????try?{
????????????ChannelFuture?future?=?bootstrap.bind(socketAddress).sync();
????????????log.info("服務(wù)器啟動(dòng)開(kāi)始監(jiān)聽(tīng)端口:?{}",?socketAddress.getPort());
????????????future.channel().closeFuture().sync();
????????}?catch?(InterruptedException?e)?{
????????????log.error("服務(wù)器開(kāi)啟失敗",?e);
????????}?finally?{
????????????//關(guān)閉主線程組
????????????bossGroup.shutdownGracefully();
????????????//關(guān)閉工作線程組
????????????workGroup.shutdownGracefully();
????????}
????}
}

ServerChannelInitializer.java：netty服務(wù)初始化器

/**
*?netty服務(wù)初始化器
**/
public?class?ServerChannelInitializer?extends?ChannelInitializer<SocketChannel>?{
????@Override
????protected?void?initChannel(SocketChannel?socketChannel)?throws?Exception?{
????????//添加編解碼
????????socketChannel.pipeline().addLast("decoder",?new?StringDecoder(CharsetUtil.UTF_8));
????????socketChannel.pipeline().addLast("encoder",?new?StringEncoder(CharsetUtil.UTF_8));
????????socketChannel.pipeline().addLast(new?NettyServerHandler());
????}
}

NettyServerHandler.java：netty服務(wù)端處理器

/**
*?netty服務(wù)端處理器
**/
@Slf4j
public?class?NettyServerHandler?extends?ChannelInboundHandlerAdapter?{
????/**
?????*?客戶(hù)端連接會(huì)觸發(fā)
?????*/
????@Override
????public?void?channelActive(ChannelHandlerContext?ctx)?throws?Exception?{
????????log.info("Channel?active......");
????}

????/**
?????*?客戶(hù)端發(fā)消息會(huì)觸發(fā)
?????*/
????@Override
????public?void?channelRead(ChannelHandlerContext?ctx,?Object?msg)?throws?Exception?{
????????log.info("服務(wù)器收到消息:?{}",?msg.toString());
????????ctx.write("你也好哦");
????????ctx.flush();
????}


????/**
?????*?發(fā)生異常觸發(fā)
?????*/
????@Override
????public?void?exceptionCaught(ChannelHandlerContext?ctx,?Throwable?cause)?throws?Exception?{
????????cause.printStackTrace();
????????ctx.close();
????}
}

RpcServerApp.java：SpringBoot啟動(dòng)類(lèi)

/**
*?啟動(dòng)類(lèi)
*
*/
@Slf4j
@SpringBootApplication(exclude?=?{DataSourceAutoConfiguration.class})
public?class?RpcServerApp?extends?SpringBootServletInitializer?{
????@Override
????protected?SpringApplicationBuilder?configure(SpringApplicationBuilder?application)?{
????????return?application.sources(RpcServerApp.class);
????}

????/**
?????*?項(xiàng)目的啟動(dòng)方法
?????*
?????*?@param?args
?????*/
????public?static?void?main(String[]?args)?{
????????SpringApplication.run(RpcServerApp.class,?args);
????????//開(kāi)啟Netty服務(wù)
????????NettyServer?nettyServer?=new??NettyServer?();
????????nettyServer.start();
????????log.info("======服務(wù)已經(jīng)啟動(dòng)========");
????}
}

客戶(hù)端部分

NettyClientUtil.java：NettyClient工具類(lèi)

/**
*?Netty客戶(hù)端
**/
@Slf4j
public?class?NettyClientUtil?{

????public?static?ResponseResult?helloNetty(String?msg)?{
????????NettyClientHandler?nettyClientHandler?=?new?NettyClientHandler();
????????EventLoopGroup?group?=?new?NioEventLoopGroup();
????????Bootstrap?bootstrap?=?new?Bootstrap()
????????????????.group(group)
????????????????//該參數(shù)的作用就是禁止使用Nagle算法，使用于小數(shù)據(jù)即時(shí)傳輸
????????????????.option(ChannelOption.TCP_NODELAY,?true)
????????????????.channel(NioSocketChannel.class)
????????????????.handler(new?ChannelInitializer<SocketChannel>()?{
????????????????????@Override
????????????????????protected?void?initChannel(SocketChannel?socketChannel)?throws?Exception?{
????????????????????????socketChannel.pipeline().addLast("decoder",?new?StringDecoder());
????????????????????????socketChannel.pipeline().addLast("encoder",?new?StringEncoder());
????????????????????????socketChannel.pipeline().addLast(nettyClientHandler);
????????????????????}
????????????????});
????????try?{
????????????ChannelFuture?future?=?bootstrap.connect("127.0.0.1",?8082).sync();
????????????log.info("客戶(hù)端發(fā)送成功....");
????????????//發(fā)送消息
????????????future.channel().writeAndFlush(msg);
????????????//?等待連接被關(guān)閉
????????????future.channel().closeFuture().sync();
????????????return?nettyClientHandler.getResponseResult();
????????}?catch?(Exception?e)?{
????????????log.error("客戶(hù)端Netty失敗",?e);
????????????throw?new?BusinessException(CouponTypeEnum.OPERATE_ERROR);
????????}?finally?{
????????????//以一種優(yōu)雅的方式進(jìn)行線程退出
????????????group.shutdownGracefully();
????????}
????}
}

NettyClientHandler.java：客戶(hù)端處理器

/**
*?客戶(hù)端處理器
**/
@Slf4j
@Setter
@Getter
public?class?NettyClientHandler?extends?ChannelInboundHandlerAdapter?{

????private?ResponseResult?responseResult;

????@Override
????public?void?channelActive(ChannelHandlerContext?ctx)?throws?Exception?{
????????log.info("客戶(hù)端Active?.....");
????}

????@Override
????public?void?channelRead(ChannelHandlerContext?ctx,?Object?msg)?throws?Exception?{
????????log.info("客戶(hù)端收到消息:?{}",?msg.toString());
????????this.responseResult?=?ResponseResult.success(msg.toString(),?CouponTypeEnum.OPERATE_SUCCESS.getCouponTypeDesc());
????????ctx.close();
????}

????@Override
????public?void?exceptionCaught(ChannelHandlerContext?ctx,?Throwable?cause)?throws?Exception?{
????????cause.printStackTrace();
????????ctx.close();
????}
}

驗(yàn)證

測(cè)試接口

@RestController
@Slf4j
public?class?UserController?{

????@PostMapping("/helloNetty")
????@MethodLogPrint
????public?ResponseResult?helloNetty(@RequestParam?String?msg)?{
????????return?NettyClientUtil.helloNetty(msg);
????}
}

訪問(wèn)測(cè)試接口

服務(wù)端打印信息

客戶(hù)端打印信息