Netty入門（二）時間服務器及客戶端

 

　　在這個例子中，我在服務器和客戶端連接被創立時發送一個消息，然后在客戶端解析收到的消息並輸出。並且，在這個項目中我使用 POJO 代替 ByteBuf 來作為傳輸對象。

一、服務器實現

1.  首先我們自定義傳輸數據對象

package com.coder.client;

import java.util.Date;

/**
 * 自定義時間數據類
 * @author Coder
 *
 */
public class Time {
    private final long value;

    public Time() {
        // 除以1000是為了使時間精確到秒
        this(System.currentTimeMillis() / 1000L);
    }

    public Time(long value) {
        this.value = value;
    }

    public long value() {
        return value;
    }

    @Override
    public String toString() {
        return new Date((value()) * 1000L).toString();
    }
}

 

 

2.  然后我們需要自定義服務器數據編碼類

package com.coder.server;

import com.coder.client.Time;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

/**
 * 服務器數據編碼類
 * @author Coder
 *
 */
public class TimeEncoderPOJO extends MessageToByteEncoder<Time> {

    // 發送數據時調用
    @Override
    protected void encode(ChannelHandlerContext ctx, Time msg, ByteBuf out) throws Exception {
        // 只傳輸當前時間，精確到秒
        out.writeInt((int)msg.value());
    }

}

 

 

3. 也需要自定義服務器的業務邏輯類，如下：

package com.coder.server;

import com.coder.client.Time;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 服務器解碼器
 * 連接建立時發送當前時間
 * @author Coder
 *
 */
public class TimeServerHandlerPOJO extends ChannelInboundHandlerAdapter {
    /**
     * 連接建立的時候並且准備進行通信時被調用
     */
    @Override
    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
        // 發送當前時間信息
        ChannelFuture f = ctx.writeAndFlush(new Time());
        // 發送完畢之后關閉 Channel
        f.addListener(ChannelFutureListener.CLOSE);
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

 

 

4. 有了上面的代碼，我們就可以實現服務器程序了，如下：

package com.coder.server;

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

public class TimeServerPOJO {
private int port;
    
    public TimeServerPOJO(int port) {
        this.port = port;
    }
    
    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();        // 用來接收進來的連接
        EventLoopGroup workerGroup = new NioEventLoopGroup();    // 用來處理已經被接收的連接
        System.out.println("准備運行端口：" + port);
        
        try {
            ServerBootstrap b = new ServerBootstrap();        // 啟動NIO服務的輔助啟動類
            b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)            // 這里告訴Channel如何接收新的連接
            .childHandler( new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    // 自定義處理類
                    // 注意添加順序
                    ch.pipeline().addLast(new TimeEncoderPOJO(),new TimeServerHandlerPOJO());
                }
            })
            .option(ChannelOption.SO_BACKLOG, 128)
            .childOption(ChannelOption.SO_KEEPALIVE, true);
            
            // 綁定端口，開始接收進來的連接
            ChannelFuture f = b.bind(port).sync();
            
            // 等待服務器socket關閉
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8080;
        new TimeServer(port).run();
    }
}

　　執行代碼后如下：

　　

　　這時候服務器在等待客戶端的連接（非阻塞）。

 

二、客戶端實現

 　　客戶端的實現與服務器類似。

1. 自定義客戶端數據解碼類

package com.coder.client;

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

public class TimeDecoderPOJO extends ByteToMessageDecoder {
    /**
     * 有新數據接收時調用
     * 為防止分包現象，先將數據存入內部緩存，到達滿足條件之后再進行解碼
     */
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if(in.readableBytes() < 4) {
            return;
        }
        
        // out添加對象則表示解碼成功
        out.add(new Time(in.readUnsignedInt()));
    }
}

 

 

2. 自定義客戶端業務邏輯類

package com.coder.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 客戶端數據處理類
 * @author Coder
 *
 */
public class TimeClientHandlerPOJO extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 直接將信息轉換成Time類型輸出即可
        Time time = (Time)msg;
        System.out.println(time);
        ctx.close();
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

 

 

3. 客戶端程序實現

 　　Netty 客戶端的通信步驟大致為：

1.  創建一個 NIO 線程組，用於處理服務器與客戶端的連接，客戶端不需要用到 boss worker。
2.  創建一個 Bootstrap 對象，配置 Netty 的一系列參數，由於客戶端 SocketChannel 沒有父親，所以不需要使用 childoption。
3.  創建一個用於實際處理數據的類ChannelInitializer，進行初始化的准備工作，比如設置接受傳出數據的字符集、格式以及實際處理數據的接口。
4.  配置服務器 IP 和端口號，建立與服務器的連接。

package com.coder.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class TimeClientPOJO {
    public static void main(String[] args) throws Exception{
        String host = "127.0.0.1";            // ip
        int port = 8080;                    // 端口
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            Bootstrap b = new Bootstrap();            // 與ServerBootstrap類似
            b.group(workerGroup);                    // 客戶端不需要boss worker
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.SO_KEEPALIVE, true);    // 客戶端的socketChannel沒有父親
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    // POJO
                    ch.pipeline().addLast(new TimeDecoderPOJO() ,new TimeClientHandlerPOJO());
                }
            });
            
            // 啟動客戶端，客戶端用connect連接
            ChannelFuture f = b.connect(host, port).sync();
            
            // 等待連接關閉
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

 

 

 

 

 三、測試

 　　先運行服務器程序，運行結果如下圖：

　　

　　然后運行客戶端程序，運行結果如下圖：

　　

　　需要注意的是，Eclipse 是可以同時運行多個 Java 程序的，可以通過點擊

　　

　　來切換不同程序的控制台輸出窗口。