Spring - Netty （整合）

時間 2019-11-07

原文原文鏈接

瘋狂創客圈，一個Java 高併發研習社羣 【博客園總入口】html

寫在前面

你們好，我是做者尼恩。目前和幾個小夥伴一塊兒，組織了一個高併發的實戰社羣【瘋狂創客圈】。正在開始高併發、億級流程的 IM 聊天程序學習和實戰，此文是： 瘋狂創客圈 Java 分佈式聊天室【億級流量】實戰系列之 -30 java

順便說明下：
本文的內容只是一個初稿、初稿，本文的知識，在《Netty Zookeeper Redis 高併發實戰》一書時，進行大篇幅的完善和更新，而且進行的源碼的升級。博客和書不同，書的內容更加系統化、全面化，更加層層升入、井井有條、更屢次的錯誤排查，請你們以書的內容爲準。
本文的最終內容，具體請參考瘋狂創客圈 傾力編著，機械工業出版社出版的《Netty Zookeeper Redis 高併發實戰》一書。面試

Spring Netty 整合實戰

瘋狂創客圈死磕Netty 系列之11

主要介紹的是SpringBoot整合Netty。在使用Netty以前，建議先了解Netty的基本原理，請參閱瘋狂創客圈。spring

這裏僅僅是使用Netty的第一步，這裏介紹一個最簡單的Demo——EchoServer，也就是回寫服務器。就是不管客戶端發啥字符串到服務器端，服務器端接收字符串後直接回寫到客戶端。bootstrap

源碼下載連接：

點擊下載數組

本篇內容綱要

環境要求springboot
Spring +netty 服務器端服務器
Spring +netty 客戶端併發
Spring讀取配置文件中的屬性值app

環境要求

JDK:：1.8
Netty:：4.0或以上(不包括5)

<java.version>1.8</java.version>
<springboot>1.5.9.RELEASE</springboot>
<netty.version>4.0.33.Final</netty.version>

Spring +netty 服務器端

回寫服務器 Echo Server 程序主要由兩部分組成：

ServerBootstrap:服務器啓動引導器。負責配置服務器端基本信息，而且完成服務器的啓動
EchoServerHandler：回寫的業務邏輯處理器

ServerBootstrap

首先是編寫服務端的啓動類，代碼中相應的註釋在寫得很詳細。主要的步驟以下：

建立一個ServerBootstrap實例
建立一個EventLoopGroup來處理各類事件，如處理連接請求，發送接收數據等。
設置本地監聽端口 InetSocketAddress( port)
設置 childHandler 來設置通道初始化類。而且在通道初始化時，加入回寫的業務邏輯處理器EchoServerHandler到服務器通道的pipeline中。childHandler 在通道初始化時，會被執行一次。
全部準備好以後調用ServerBootstrap.bind() 方法綁定 Server

不過須要注意的是，在不使用Spring的環境中，是經過main方法直接啓動服務端，所以是直接new一個處理器echoServerHandler 對象。而在和Spring 整合以後，咱們須要將 echoServerHandler 處理器交給springBoot去管理。

ServerBootstrap 代碼以下：

@Service("EchoServer") public class EchoServer { // 服務器端口 @Value("${server.port}") private int port; // 經過nio方式來接收鏈接和處理鏈接 private static EventLoopGroup boss = new NioEventLoopGroup(); private static EventLoopGroup work = new NioEventLoopGroup(); // 啓動引導器 private static ServerBootstrap b = new ServerBootstrap(); @Autowired private EchoServerHandler echoServerHandler; public void run() { try { b.group(boss, work); // 設置nio類型的channel b.channel(NioServerSocketChannel.class); // 設置監聽端口 b.localAddress(new InetSocketAddress(port)); // 設置通道初始化 b.childHandler(new ChannelInitializer<SocketChannel>() { //有鏈接到達時會建立一個channel protected void initChannel(SocketChannel ch) throws Exception { // pipeline管理channel中的Handler // 在channel隊列中添加一個handler來處理業務 ch.pipeline().addLast("echoServerHandler",echoServerHandler); } }); // 配置完成，開始綁定server // 經過調用sync同步方法阻塞直到綁定成功 ChannelFuture f = b.bind().sync(); System.out.println(EchoServer.class.getName() + " started and listen on " + f.channel().localAddress()); // 監聽服務器關閉事件 // 應用程序會一直等待，直到channel關閉 f.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { // 關閉EventLoopGroup，釋放掉全部資源包括建立的線程 work.shutdownGracefully(); boss.shutdownGracefully(); } } }

業務邏輯ServerHandler：

要想處理接收到的數據，咱們必須繼承ChannelInboundHandlerAdapter接口，重寫裏面的channelRead方法，每當有數據到達，此方法就會被調用（通常是Byte類型數組），咱們就在這裏寫咱們的業務邏輯：

@Service("echoServerHandler") public class EchoServerHandler extends ChannelInboundHandlerAdapter { /** * 創建鏈接時，發送一條消息 */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("鏈接的客戶端地址:" + ctx.channel().remoteAddress()); super.channelActive(ctx); } public void channelRead(ChannelHandlerContext ctx, Object msg) { try { System.out.println("server received data :" + msg); ctx.write(msg);//寫回數據， } finally { ReferenceCountUtil.release(msg); } } public void channelReadComplete(ChannelHandlerContext ctx) { //flush掉全部寫回的數據 ctx.writeAndFlush(Unpooled.EMPTY_BUFFER) .addListener(ChannelFutureListener.CLOSE); //當flush完成後關閉channel } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { //捕捉異常信息 cause.printStackTrace(); //出現異常時關閉channel ctx.close(); } }

關於異常處理：

咱們在上面程序中也重寫了exceptionCaught方法，這裏就是對當異常出現時的處理。

Spring +netty 客戶端

EchoClient 扮演以下角色：

鏈接到Server
向Server寫數據,
等待Server返回數據

回寫客戶端程序EchoClient 主要由兩部分組成：

Bootstrap:客戶端啓動引導器。負責配置客戶端基本信息，而且完成客戶端的啓動
EchoClientHandler ：客戶端業務邏輯處理器

EchoClient Bootstrap的過程：

和Server端相似，只不過Client端要同時指定鏈接主機的IP和Port。

建立一個Bootstrap實例
建立一個EventLoopGroup 來處理各類事件，如處理連接請求，發送接收數據等。
定義須要鏈接到的遠程服務器的InetSocketAddress,包含了IP+端口
設置 childHandler 來設置通道初始化類。而且在通道初始化時，加入客戶端的業務邏輯處理器echoClientHandler 到服務器通道的pipeline中。當鏈接完成以後，childHandler 會被執行一次。
全部準備好以後調用 ServerBootstrap.connect() 方法鏈接Server

EchoClient Bootstrap的代碼：

@Service("EchoClient") public class EchoClient { // 服務器ip地址 @Value("${server.ip}") private String host; // 服務器端口 @Value("${server.port}") private int port; // 經過nio方式來接收鏈接和處理鏈接 private EventLoopGroup group = new NioEventLoopGroup(); @Autowired private EchoClientHandler echoClientHandler; /** * 惟一標記 */ private boolean initFalg = true; /** * 客戶端的是Bootstrap，服務端的則是 ServerBootstrap。 * 都是AbstractBootstrap的子類。 **/ public void run() { doConnect(new Bootstrap(), group); } /** * 重連 */ public void doConnect(Bootstrap bootstrap, EventLoopGroup eventLoopGroup) { ChannelFuture f = null; try { if (bootstrap != null) { bootstrap.group(eventLoopGroup); bootstrap.channel(NioSocketChannel.class); bootstrap.option(ChannelOption.SO_KEEPALIVE, true); bootstrap.remoteAddress(host, port); // 設置通道初始化 bootstrap.handler( new ChannelInitializer<SocketChannel>() { public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(echoClientHandler); } } ); f = bootstrap.connect().addListener((ChannelFuture futureListener) -> { final EventLoop eventLoop = futureListener.channel().eventLoop(); if (!futureListener.isSuccess()) { System.out.println("與服務端斷開鏈接!在10s以後準備嘗試重連!"); eventLoop.schedule(() -> doConnect(new Bootstrap(), eventLoop), 10, TimeUnit.SECONDS); } }); if (initFalg) { System.out.println("EchoClient客戶端鏈接成功!"); initFalg = false; } // 阻塞 f.channel().closeFuture().sync(); } } catch (Exception e) { System.out.println("客戶端鏈接失敗!" + e.getMessage()); } } }

EchoClientHandler 客戶端業務邏輯處理器

要想處理接收到的數據，咱們必須繼承ChannelInboundHandlerAdapter基類，重寫裏面的channelRead方法，每當有數據到達，此方法就會被調用（通常是Byte類型數組），咱們就在這裏寫咱們的業務邏輯：

@Service("echoClientHandler") public class EchoClientHandler extends ChannelInboundHandlerAdapter { /** * 此方法會在鏈接到服務器後被調用 */ public void channelActive(ChannelHandlerContext ctx) { ctx.write(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8)); } /** * 業務邏輯處理 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // 若是不是protobuf類型的數據 if (!(msg instanceof ByteBuf)) { System.out.println("未知數據!" + msg); return; } try { ByteBuf in = (ByteBuf) msg; System.out.println("Client received: " + ByteBufUtil.hexDump(in.readBytes(in.readableBytes()))); } catch (Exception e) { e.printStackTrace(); } finally { ReferenceCountUtil.release(msg); } } /** * 捕捉到異常 */ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); } }

除了繼承ChannelInboundHandlerAdapter基類，咱們的業務Handler還能夠繼承 SimpleChannelInboundHandler 基類。

那麼這兩個有什麼區別呢？

SimpleChannelInboundHandler在接收到數據後會自動release掉數據佔用的Bytebuffer資源(自動調用Bytebuffer.release())。若是在channelRead方法返回前尚未寫完數據，也就是當不能讓它自動release時，就不能繼承 SimpleChannelInboundHandler 基類。而繼承ChannelInboundHandlerAdapter則不會自動釋放，須要手動調用ReferenceCountUtil.release()等方法進行釋放。

SimpleChannelInboundHandler還有一個好處，能夠在泛型參數中，能夠直接指定好傳輸的數據格式。因此繼承該類，在處理數據時，不須要判斷數據格式。而繼承ChannelInboundHandlerAdapter則須要進行數據格式的判斷和轉換。
推薦在服務端去繼承ChannelInboundHandlerAdapter，建議手動進行釋放，防止數據未處理完就自動釋放了。

Spring讀取配置文件中的屬性值

在Netty 的程序中，通常須要用到服務器ip和端口，最好的方式是放在配置文件中，方便修改。

Spring Boot 默認的配置文件名稱爲 application.properties，SpringApplication將從如下位置加載此文件：

當前目錄下的/config子目錄，
當前目錄
一個classpath下的/config包
classpath 根路徑（root）

通常狀況下，工程在編譯以後，application.properties 放在classpath 根路徑下。

配置文件 application.properties

#端口號 server.port=8081 IPserver.ip=127.0.0.1

注意：文件名字不能錯哦，是application.properties

關聯配置項到類屬性

在類域屬性上經過@Value("${配置項}")指定關聯屬性，Spring Application會自動加載。

public class EchoServer { // 服務器端口 @Value("${server.port}") private int port; //... }

啓動配置項自動掃描

使用 @Configuration、@EnableAutoConfiguration 啓動配置項的自動掃描。

//自動加載配置信息 @Configuration @EnableAutoConfiguration //使包路徑下帶有@Value的註解自動注入 //使包路徑下帶有@Autowired的類能夠自動注入 @ComponentScan("com.crazymakercircle.nettydemo.server") @SpringBootApplication public class ServerApp { // ........ }