如何在SpringBoot中，使用Netty實(shí)現(xiàn)遠(yuǎn)程調(diào)用？
前言
眾所周知我們?cè)谶M(jìn)行網(wǎng)絡(luò)連接的時(shí)候，建立套接字連接是一個(gè)非常消耗性能的事情，特別是在分布式的情況下，用線程池去保持多個(gè)客戶端連接，是一種非常消耗線程的行為。那么我們?cè)撏ㄟ^(guò)什么技術(shù)去解決上述的問(wèn)題呢，那么就不得不提一個(gè)網(wǎng)絡(luò)連接的利器——Netty.
Netty
Netty是一個(gè)NIO客戶端服務(wù)器框架：
它可快速輕松地開(kāi)發(fā)網(wǎng)絡(luò)應(yīng)用程序，例如協(xié)議服務(wù)器和客戶端。
它極大地簡(jiǎn)化和簡(jiǎn)化了網(wǎng)絡(luò)編程，例如TCP和UDP套接字服務(wù)器。
NIO是一種非阻塞IO ，它具有以下的特點(diǎn)
單線程可以連接多個(gè)客戶端。
選擇器可以實(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模式，從而解決客戶端請(qǐng)求在服務(wù)端不完整的問(wèn)題。
netty模型reactor模式
在selector的基礎(chǔ)上解決了半包問(wèn)題。
上圖，簡(jiǎn)單地可以描述為"boss接活，讓work干"：manReactor用來(lái)接收請(qǐng)求（會(huì)與客戶端進(jìn)行握手驗(yàn)證），而subReactor用來(lái)處理請(qǐng)求（不與客戶端直接連接）。
SpringBoot使用Netty實(shí)現(xiàn)遠(yuǎn)程調(diào)用
maven依賴
<!--lombok-->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.2</version>
<optional>true</optional>
</dependency>
<!--netty-->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.17.Final</version>
</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 {
/**
* 客戶端連接會(huì)觸發(fā)
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("Channel active......");
}
/**
* 客戶端發(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)類
/**
* 啟動(dòng)類
*
*/
@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)========");
}
}客戶端部分
NettyClientUtil.java：NettyClient工具類
/**
* Netty客戶端
**/
@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("客戶端發(fā)送成功....");
//發(fā)送消息
future.channel().writeAndFlush(msg);
// 等待連接被關(guān)閉
future.channel().closeFuture().sync();
return nettyClientHandler.getResponseResult();
} catch (Exception e) {
log.error("客戶端Netty失敗", e);
throw new BusinessException(CouponTypeEnum.OPERATE_ERROR);
} finally {
//以一種優(yōu)雅的方式進(jìn)行線程退出
group.shutdownGracefully();
}
}
}NettyClientHandler.java：客戶端處理器
/**
* 客戶端處理器
**/
@Slf4j
@Setter
@Getter
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
private ResponseResult responseResult;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("客戶端Active .....");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("客戶端收到消息: {}", 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ù)端打印信息
客戶端打印信息
作者：程序員DSC阿里
出處：blog.csdn.net/m0_57473261/article/details/117893030
關(guān)注GitHub今日熱榜，專注挖掘好用的開(kāi)發(fā)工具，致力于分享優(yōu)質(zhì)高效的工具、資源、插件等，助力開(kāi)發(fā)者成長(zhǎng)！
點(diǎn)個(gè)在看，你最好看