Netty(一)创建简单的Echo Server & Client

Netty简介

Netty的核心组件:

  • Channel
  • 回调
  • Future
  • 事件和ChannelHandler

Channel
Channel可以看做是传入(入站)或者传出(出站)数据的载体,,可以被打开或者关闭,连接或者断开连接;

回调
一个回调其实就是一个方法,一个指向已经被提供给另外一个方法的方法的引用。这使得接受回调的方法可以在适当的时候调用前者。当一个回调被触发时,相关的事件可以被一个interface-ChannelHandler的实现处理。

Future
Future提供了另一种在操作完成时通知应用程序的方式。这个对象可以看做是一个异步操作的结果的占位符,它将在未来的某个时刻完成,并提供对其结果的访问。
ChannelFuture提供了额外几个方法,使得我们能够注册一个或者多个ChannelFutureListener实例。由ChannelFutureListener实例提供的通知机制消除了手动检查对应的操作是否完成的必要。

事件和 ChannelHandler
Netty使用不同的事件来通知我们状态的改变或者是操作的状态。这使得我们能够基于已经发生的事件来触发适当的动作。每个事件都可以被分发给ChannelHandler类中的某个用户实现的方法。

创建Maven工程

整个项目的目录结构如下:

Netty(一)创建简单的Echo Server & Client

echo-parent的 pom文件

<?xml version="1.0" encoding="UTF-8"?>
<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.nasuf.echo</groupId>
    <artifactId>echo-parent</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>

    <modules>
        <module>../echo-client</module>
        <module>../echo-server</module>
    </modules>

    <properties>
        <echo-server.hostname>localhost</echo-server.hostname>
        <echo-server.port>9999</echo-server.port>
    </properties>

    <dependencies>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId> <!-- Use 'netty-all' for 4.0 or above -->
            <version>4.1.10.Final</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
            </plugin>
            <plugin>
                <artifactId>maven-failsafe-plugin</artifactId>
            </plugin>
            <plugin>
                <artifactId>maven-surefire-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

echo-server的 pom文件

<?xml version="1.0" encoding="UTF-8"?>
<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">
    <parent>
        <artifactId>echo-parent</artifactId>
        <groupId>com.nasuf.echo</groupId>
        <version>1.0-SNAPSHOT</version>
        <relativePath>../echo-parent/pom.xml</relativePath>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>echo-server</artifactId>

    <build>
        <plugins>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>run-server</id>
                        <goals>
                            <goal>java</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <mainClass>com.echo.server.handler.EchoServer</mainClass>
                    <arguments>
                        <argument>${echo-server.port}</argument>
                    </arguments>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

echo-client的 pom文件

<?xml version="1.0" encoding="UTF-8"?>
<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">
    <parent>
        <artifactId>echo-parent</artifactId>
        <groupId>com.nasuf.echo</groupId>
        <version>1.0-SNAPSHOT</version>
        <relativePath>../echo-parent/pom.xml</relativePath>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>echo-client</artifactId>

    <build>
        <plugins>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>run-server</id>
                        <goals>
                            <goal>java</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <mainClass>com.echo.client.handler.EchoClient</mainClass>
                    <arguments>
                        <argument>${echo-server.hostname}</argument>
                        <argument>${echo-server.port}</argument>
                    </arguments>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

然后来构建 echo-clientecho-server

EchoServer

EchoServerHandler

package com.echo.server.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

// @Sharable标示一个ChannelHandler可以被多个Channel安全共享
@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf in = (ByteBuf) msg;
        System.out.println(
                "Server received: " + in.toString(CharsetUtil.UTF_8));
        // 将接收到的消息写给发送者,即客户端,而不冲刷出站消息        
        ctx.write(in);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将未决消息冲刷到远程节点,并且关闭该Channel
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
            .addListener(ChannelFutureListener.CLOSE);
    }

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

}

所谓“未决消息(pending message)”指的是目前暂存于ChannelOutboundBuffer中的消息,在下一次调用flush()或者writeAndFlush()方法时将会尝试写出到套接字。

EchoServer

package com.echo.server.handler;

import java.net.InetSocketAddress;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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 EchoServer {
    
    private final int port;
    
    public EchoServer(int port) {
        this.port = port;
    }
    
    public void start() throws Exception {
        final EchoServerHandler serverHandler = new EchoServerHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)
                .channel(NioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(port))
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(serverHandler);
                    }
                });
            // 此处绑定服务器,并等待绑定完成。对sync()方法的调用将导致当前Thread阻塞,直到绑定完成
            ChannelFuture f = b.bind().sync();
            // 由于调用了sync()方法,程序将会阻塞等待,直到服务器的Channel关闭
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();
        }
    }
    
    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            System.err.println(
                    "Usage: " + EchoServer.class.getSimpleName() + " <port>"
                    );
            return;
        }
        int port = Integer.parseInt(args[0]);
        new EchoServer(port).start();
    }

}

总结:

  • EchoServerHandler实现了业务逻辑
  • main()方法引导了服务器;

引导过程如下:

  • 创建一个ServerBoostrap的实例以引导和绑定服务器;
  • 创建并分配一个NioEventLoopGroup实例以进行事件的处理,如接受新连接以及读、写数据;
  • 指定服务器绑定本地的InetSocketAddress;
  • 使用一个EchoServerHandler的实例初始化每一个新Channel;
  • 调用ServerBootstrap.bind()方法以绑定服务器;

EchoClient

Echo客户端将会:

  • 连接到服务器;
  • 发送一个或多个消息;
  • 对于每个消息,等待并接收从服务器返回的相同的消息;
  • 关闭连接

EchoClientHandler

package com.echo.client.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

@Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        System.out.println("Client received: " + in.toString(CharsetUtil.UTF_8));
    }

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

EchoCleint

package com.echo.client.handler;

import java.net.InetSocketAddress;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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 EchoClient {
    
    private final String host;
    private final int port;
    
    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println(
                    "Usage: " + EchoClient.class.getSimpleName() +
                    " <host> <port>");
            return;
        }
        String host = args[0];
        int port = Integer.parseInt(args[1]);
        new EchoClient(host, port).start();
    }
    
    public void start() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .remoteAddress(new InetSocketAddress(host, port))
             .handler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new EchoClientHandler());
                }
                 
             });
            ChannelFuture f = b.connect().sync();
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();
        }
    }

}

总结:

  • 为初始化客户端,创建了一个Bootstrap实例;
  • 为进行事件处理分配了一个NioEventLoopGroup实例,其中事件处理包括创建新的连接以及处理入站和出站数据;
  • 为服务器连接创建了一个InetSocketAddress实例;
  • 当连接被建立时,一个EchoClientHandler实例会被安装到(该Channel的)ChannelPipeline中;
  • 在一切都设置完成后,调用Bootstrap.connect()方法连接到远程节点;

运行测试

进入到echo-parent目录下执行:

mvn clean package

然后分别在服务端和客户端执行:

mvn exec:java

可以在服务端看到:

$ mvn exec:java
[INFO] Scanning for projects...
[INFO] 
[INFO] ------------------------------------------------------------------------
[INFO] Building echo-server 1.0-SNAPSHOT
[INFO] ------------------------------------------------------------------------
[INFO] 
[INFO] --- exec-maven-plugin:1.6.0:java (default-cli) @ echo-server ---

在客户端看到:

$ mvn exec:java
[INFO] Scanning for projects...
[INFO] 
[INFO] ------------------------------------------------------------------------
[INFO] Building echo-client 1.0-SNAPSHOT
[INFO] ------------------------------------------------------------------------
[INFO] 
[INFO] --- exec-maven-plugin:1.6.0:java (default-cli) @ echo-client ---
Client received: Netty rocks!
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 3.903 s
[INFO] Finished at: 2018-11-18T20:54:08+08:00
[INFO] Final Memory: 11M/309M
[INFO] ------------------------------------------------------------------------

客户端执行完毕并退出,回到服务端窗口可以看到信息如下:

Server received: Netty rocks!
《Netty实战》第二章

相关推荐