netty搭建web聊天室（1）

Netty · 發表 2019-01-21 16:33:42

摘要：之前一直在搞前端的東西，都快忘了自己是個java開發。其實還有好多java方面的東西沒搞過，突然瞭解到netty，覺得有必要學一學。介紹 Netty是由JBOSS提供的一個java開源框架。Netty提供非同步的、事件驅動的網路應用程式框架和工具，用以快速開發高效能、高可靠性的網路伺...

之前一直在搞前端的東西，都快忘了自己是個java開發。其實還有好多java方面的東西沒搞過，突然瞭解到netty，覺得有必要學一學。

介紹

Netty是由JBOSS提供的一個java開源框架。Netty提供非同步的、事件驅動的網路應用程式框架和工具，用以快速開發高效能、高可靠性的網路伺服器和客戶端程式。

也就是說，Netty 是一個基於NIO的客戶、伺服器端程式設計框架，使用Netty 可以確保你快速和簡單的開發出一個網路應用，例如實現了某種協議的客戶、服務端應用。Netty相當於簡化和流線化了網路應用的程式設計開發過程，例如：基於TCP和UDP的socket服務開發。

一些IO概念

NIO （non-blocking IO）非阻塞
BIO （blocking IO）阻塞

以上兩種又可分為同步和非同步，即同步阻塞，同步非阻塞，非同步阻塞，非同步非阻塞。

阻塞：資料沒來，啥都不做，直到資料來了，才進行下一步的處理。
非阻塞：資料沒來，程序就不停的去檢測資料，直到資料來。

至於這塊的詳細概念，大家可以自行百度學習。總之，netty處理io很高效，不需要你擔心。

netty結構

可以看出它支援的網路傳輸協議，以及容器支援，安全支援,io.

工作流程：

所有客戶端的連線交給住主執行緒去管理，響應客戶端的訊息交給從執行緒去處理，整個執行緒池由netty負責。

搭建服務

建立maven工程引入最新的依賴

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.mike</groupId>
<artifactId>netty</artifactId>
<version>0.0.1-SNAPSHOT</version>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties>

<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.32.Final</version>
</dependency>
</dependencies>
</project>

建立訊息處理器

package netty;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 
 */
public class ChatHandler extends SimpleChannelInboundHandler{

public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

/**
* 每當從服務端收到新的客戶端連線時，客戶端的 Channel 存入ChannelGroup列表中，並通知列表中的其他客戶端 Channel
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
for (Channel channel : channels) {
channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n");
}
channels.add(ctx.channel());
}

/**
* 每當從服務端收到客戶端斷開時，客戶端的 Channel 移除 ChannelGroup 列表中，並通知列表中的其他客戶端 Channel
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
for (Channel channel : channels) {
channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 離開\n");
}
channels.remove(ctx.channel());
}

/**
* 會話建立時
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)
Channel incoming = ctx.channel();
System.out.println("ChatClient:"+incoming.remoteAddress()+"線上");
}

/**
* 會話結束時
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)
Channel incoming = ctx.channel();
System.out.println("ChatClient:"+incoming.remoteAddress()+"掉線");
}

/**
* 出現異常
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (7)
Channel incoming = ctx.channel();
System.out.println("ChatClient:"+incoming.remoteAddress()+"異常");
// 當出現異常就關閉連線
cause.printStackTrace();
ctx.close();
}

/**
* 讀取客戶端傳送的訊息，並將資訊轉發給其他客戶端的 Channel。
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, Objectrequest) throws Exception {
FullHttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1,HttpResponseStatus.OK , Unpooled.wrappedBuffer("Hello netty"
.getBytes()));
response.headers().set("Content-Type", "text/plain");
response.headers().set("Content-Length", response.content().readableBytes());
response.headers().set("connection", HttpHeaderValues.KEEP_ALIVE);
ctx.writeAndFlush(response);

}

}

這裡面其實只需要重寫channelRead0 方法就可以了，其他是它的生命週期的方法，可以用來做日至記錄。我們在讀取訊息後，往channel裡寫入了一個http的response。

初始化我們的訊息處理器

package netty;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;

/**
 * 用來增加多個的處理類到 ChannelPipeline 上，包括編碼、解碼、SimpleChatServerHandler 等。
 */
public class ChatServerInitializer extends ChannelInitializer<SocketChannel>{

@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("HttpResponseEncoder",new HttpResponseEncoder());
pipeline.addLast("HttpRequestDecoder",new HttpRequestDecoder());
pipeline.addLast("chathandler", new ChatHandler());

System.out.println("ChatClient:"+ch.remoteAddress() +"連線上");

}

}

這個pipeline可以理解為netty的攔截器，每個訊息進來，經過各個攔截器的處理。我們需要響應http訊息，所以加入了響應編碼以及請求解碼，最後加上了我們的自定義處理器。這裡面有很多處理器，netty以及幫你定義好的。

服務啟動類

package netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * The class ChatServer
 */
public class ChatServer {
private int port;

public ChatServer(int port) {
this.port = port;
}

public void run() throws Exception {

EventLoopGroup bossGroup = new NioEventLoopGroup(); 
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); 
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) 
.childHandler(new ChatServerInitializer())
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); 

System.out.println("ChatServer 啟動了");

// 繫結埠，開始接收進來的連線
ChannelFuture f = b.bind(port).sync(); // (7)

// 等待伺服器socket 關閉 。
// 在這個例子中，這不會發生，但你可以優雅地關閉你的伺服器。
f.channel().closeFuture().sync();

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

System.out.println("ChatServer 關閉了");
}
}

public static void main(String[] args) throws Exception {
new ChatServer(8090).run();

}
}

這個啟動類就是按照上面那個結構圖來的，新增兩個執行緒組，設定channel，新增訊息處理器，配置一些選項option。