`
spdx4046
  • 浏览: 46980 次
  • 性别: Icon_minigender_1
  • 来自: 南京
社区版块
存档分类
最新评论

Netty使用初步

阅读更多
1、简介

Java1.4提供了NIO使开发者可以使用Java编写高性能的服务端程序,但使用原生的NIO API就像Linux C中网络编程一样,还是需要做IO处理、协议处理等低层次工作。所以,就像C服务端程序大量使用libevent作为网络应用框架一样,Java社区也不断涌现出基于NIO的网络应用框架。在这其中,Jboss出品的Netty就是个中翘楚。Netty是个异步的事件驱动网络应用框架,具有高性能、高扩展性等特性。Netty提供了统一的底层协议接口,使得开发者从底层的网络协议(比如TCP/IP、UDP)中解脱出来。就使用来说,开发者只要参考 Netty提供的若干例子和它的指南文档,就可以放手开发基于Netty的服务端程序了。
在Java社区,最知名的开源Java NIO框架要属Mina和Netty,而且两者渊源颇多,对两者的比较自然不少。实际上,Netty的作者原来就是Mina作者之一,所以可以想到,Netty和Mina在设计理念上会有很多共同点。我对Mina没什么研究,但其作者介绍,Netty的设计对开发者有更友好的扩展性,并且性能方面要优于Mina,而Netty完善的文档也很吸引人。所以,如果你在寻找Java NIO框架,Netty是个很不错的选择。本文的内容就是围绕一个demo介绍使用Netty的点点滴滴。
2、服务端程序
2.1、ChannelHandler
服务端程序通常的处理过程是:解码请求数据、业务逻辑处理、编码响应。从框架角度来说,可以提供3个接口来控制并调度该处理过程;从更通用的角度来说,并不特化处理其中的每一步,而把每一步当做过滤器链中的一环,这也是Netty的做法。Netty对请求处理过程实现了过滤器链模式(ChannelPipeline),每个过滤器实现了ChannelHandler接口。Netty中有两种请求事件流类型也做了细分:
1)downstream event:其对应的ChannelHandler子接口是ChannelDownstreamHandler。downstream event是说从头到尾执行ChannelPipeline中的ChannelDownstreamHandler,这一过程相当于向外发送数据的过程。 downstream event有:”write”、”bind”、”unbind”、 “connect”、 “disconnect”、”close”。
2)upstream event:其对应的ChannelHandler子接口是ChannelUpstreamHandler。upstream event处理的事件方向和downstream event相反,这一过程相当于接收处理外来请求的过程。upstream event有:”messageReceived”、 “exceptionCaught”、”channelOpen”、”channelClosed”、 “channelBound”、”channelUnbound”、 “channelConnected”、”writeComplete”、”channelDisconnected”、”channelInterestChanged”。
Netty中有个注释@interface ChannelPipelineCoverage,它表示被注释的ChannelHandler是否能添加到多个ChannelPipeline中,其可选的值是”all”和”one”。”all”表示ChannelHandler是无状态的,可被多个ChannelPipeline共享,而”one”表示ChannelHandler只作用于单个ChannelPipeline中。但ChannelPipelineCoverage只是个注释而已,并没有实际的检查作用。对于ChannelHandler是”all”还是”one”,还是根据逻辑需要而定。比如,像解码请求handler,因为可能解码的数据不完整,需要等待下一次读事件来了之后再继续解析,所以解码请求handler就需要是”one”的(否则多个Channel共享数据就乱了)。而像业务逻辑处理hanlder通常是”all”的。
下面以一个简单的例子说明如何编写“解码请求数据、业务逻辑处理、编码响应”这一过程中涉及的ChannelHandler。该例子实现的协议格式很简单,请求和响应流中头4个字节表示后面跟的内容长度,根据该长度可得到内容体。
首先看下解码器的实现:
public class MessageDecoder extends FrameDecoder {

    @Override
    protected Object decode(
            ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
        if (buffer.readableBytes() < 4) {
            return null;//(1)
        }
        int dataLength = buffer.getInt(buffer.readerIndex());
        if (buffer.readableBytes() < dataLength + 4) {
            return null;//(2)
        }

        buffer.skipBytes(4);//(3)
        byte[] decoded = new byte[dataLength];
        buffer.readBytes(decoded);
        String msg = new String(decoded);//(4)
        return msg;
    }
}
MessageDecoder继承自FrameDecoder,FrameDecoder是Netty codec包中的辅助类,它是个ChannelUpstreamHandler,decode方法是FrameDecoder子类需要实现的。在上面的代码中,有:
(1)检查ChannelBuffer中的字节数,如果ChannelBuffer可读的字节数少于4,则返回null等待下次读事件。
(2)继续检查ChannelBuffer中的字节数,如果ChannelBuffer可读的字节数少于dataLength + 4,则返回null等待下次读事件。
(3)越过dataLength的字节。
(4)构造解码的字符串返回。
@ChannelPipelineCoverage("all")
public class MessageServerHandler extends SimpleChannelUpstreamHandler {

    private static final Logger logger = Logger.getLogger(
            MessageServerHandler.class.getName());

    @Override
    public void messageReceived(
            ChannelHandlerContext ctx, MessageEvent e) {
        if (!(e.getMessage() instanceof String)) {
            return;//(1)
        }
        String msg = (String) e.getMessage();
        System.err.println("got msg:"+msg);
        e.getChannel().write(msg);//(2)
    }

    @Override
    public void exceptionCaught(
            ChannelHandlerContext ctx, ExceptionEvent e) {
        logger.log(
                Level.WARNING,
                "Unexpected exception from downstream.",
                e.getCause());
        e.getChannel().close();
    }
}
MessageServerHandler是服务端业务处理handler,其继承自SimpleChannelUpstreamHandler,并主要实现messageReceived事件。关于该类,有如下注解:
(1)该upstream事件流中,首先经过MessageDecoder,其会将decode返回的解码后的数据构造成 MessageEvent.getMessage(),所以在handler上下文关系中,MessageEvent.getMessage()并不一定都返回ChannelBuffer类型的数据。
(2)MessageServerHandler只是简单的将得到的msg再写回给客户端。e.getChannel().write(msg);操作将触发DownstreamMessageEvent事件,也就是调用下面的MessageEncoder将编码的数据返回给客户端。
@ChannelPipelineCoverage("all")
public class MessageEncoder extends OneToOneEncoder {

    @Override
    protected Object encode(
            ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
        if (!(msg instanceof String)) {
            return msg;//(1)
        }

        String res = (String)msg;
        byte[] data = res.getBytes();
        int dataLength = data.length;
        ChannelBuffer buf = ChannelBuffers.dynamicBuffer();//(2)
        buf.writeInt(dataLength);
        buf.writeBytes(data);
        return buf;//(3)
    }
}
MessageEncoder是个ChannelDownstreamHandler。对该类的注解如下:
(1)如果编码的msg不是合法类型,就直接返回该msg,之后OneToOneEncoder会调用 ctx.sendDownstream(evt);来调用下一个ChannelDownstreamHandler。对于该例子来说,这种情况是不应该出现的。
(2)开发者创建ChannelBuffer的用武之地就是这儿了,通常使用dynamicBuffer即可,表示得到的ChannelBuffer可动态增加大小。
(3)返回编码后的ChannelBuffer之后,OneToOneEncoder会调用Channels.write将数据写回客户端。
2.2、MessageServerPipelineFactory
创建了3个ChannelHandler,需要将他们注册到ChannelPipeline,而ChannelPipeline又是和Channel对应的(是全局单例还是每个Channel对应一个ChannelPipeline实例依赖于实现)。可以实现ChannelPipeline的工厂接口 ChannelPipelineFactory实现该目的。MessageServerPipelineFactory的代码如下:
public class MessageServerPipelineFactory implements
        ChannelPipelineFactory {

    public ChannelPipeline getPipeline() throws Exception {
        ChannelPipeline pipeline = pipeline();

        pipeline.addLast("decoder", new MessageDecoder());
        pipeline.addLast("encoder", new MessageEncoder());
        pipeline.addLast("handler", new MessageServerHandler());

        return pipeline;
    }
}
2.3、MessageServer
服务端程序就剩下启动代码了,使用Netty的ServerBootstrap三下五除二完成之。
public class MessageServer {

    public static void main(String[] args) throws Exception {
        // Configure the server.
        ServerBootstrap bootstrap = new ServerBootstrap(
                new NioServerSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool()));

        // Set up the default event pipeline.
        bootstrap.setPipelineFactory(new MessageServerPipelineFactory());

        // Bind and start to accept incoming connections.
        bootstrap.bind(new InetSocketAddress(8080));

    }
}
稍加补充的是,该Server程序并不完整,它没有处理关闭时的资源释放,尽管暴力的来看并不一定需要做这样的善后工作。
3、客户端程序
客户端程序和服务端程序处理模型上是很相似的,这里还是付上代码并作简要说明。
3.1、 ChannelHandler
客户端是先发送数据到服务端(downstream事件流),然后是处理从服务端接收的数据(upstream事件流)。这里有个问题是,怎么把需要发送的数据送到downstream事件流里呢?这就用到了ChannelUpstreamHandler的channelConnected事件了。实现的 MessageClientHandler代码如下:
@ChannelPipelineCoverage("all")
public class MessageClientHandler extends SimpleChannelUpstreamHandler {

    private static final Logger logger = Logger.getLogger(
            MessageClientHandler.class.getName());


    @Override
    public void channelConnected(
            ChannelHandlerContext ctx, ChannelStateEvent e) {
        String message = "hello kafka0102";
        e.getChannel().write(message);
    }

    @Override
    public void messageReceived(
            ChannelHandlerContext ctx, MessageEvent e) {
        // Send back the received message to the remote peer.
        System.err.println("messageReceived send message "+e.getMessage());
        try {
            Thread.sleep(1000*3);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        e.getChannel().write(e.getMessage());
    }

    @Override
    public void exceptionCaught(
            ChannelHandlerContext ctx, ExceptionEvent e) {
        // Close the connection when an exception is raised.
        logger.log(
                Level.WARNING,
                "Unexpected exception from downstream.",
                e.getCause());
        e.getChannel().close();
    }
}
对于编码和解码Handler,复用MessageEncoder和MessageDecoder即可。
3.2、 MessageClientPipelineFactory
    public class MessageClientPipelineFactory implements
        ChannelPipelineFactory {

    public ChannelPipeline getPipeline() throws Exception {
        ChannelPipeline pipeline = pipeline();

        pipeline.addLast("decoder", new MessageDecoder());
        pipeline.addLast("encoder", new MessageEncoder());
        pipeline.addLast("handler", new MessageClientHandler());

        return pipeline;
    }
}
3.3、MessageClient
public class MessageClient {

    public static void main(String[] args) throws Exception {
        // Parse options.
        String host = "127.0.0.1";
        int port = 8080;
        // Configure the client.
        ClientBootstrap bootstrap = new ClientBootstrap(
                new NioClientSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool()));
        // Set up the event pipeline factory.
        bootstrap.setPipelineFactory(new MessageClientPipelineFactory());
        // Start the connection attempt.
        ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
        // Wait until the connection is closed or the connection attempt fails.
        future.getChannel().getCloseFuture().awaitUninterruptibly();
        // Shut down thread pools to exit.
        bootstrap.releaseExternalResources();
    }
}
在写客户端例子时,我想像的代码并不是这样的,对客户端的代码我也没做过多的研究,所以也可能没有找到更好的解决方案。在上面的例子中,bootstrap.connect方法中会触发实际的连接操作,接着触发 MessageClientHandler.channelConnected,使整个过程运转起来。但是,我想要的是一个连接池,并且如何写数据也不应该在channelConnected中,这样对于动态的数据,只能在构造函数中传递需要写的数据了。但到现在,我还不清楚如何将连接池和 ChannelPipeline有效的结合起来。或许,这样的需求可以跨过Netty来实现。
4、总结
关于Netty的初步使用,尚且总结到这里。关于这篇文章,写得断断续续,以至于到后来我都没兴趣把内容都整理出来。当然,这多少也是因为我是先整理 Netty原理方面的东西所致。我也只能卑微的期望,该文对Netty入门者会有些许帮助。

=============================== 华丽的终止符 ================================

分享到:
评论

相关推荐

    Netty相关文档

    在压缩包中的"Netty使用初步.doc"可能是对Netty的入门教程,涵盖基础概念如Channel、EventLoop、ByteBuf等,以及如何创建服务器和客户端,处理各种网络事件等。通常,这样的文档会引导读者逐步实现一个简单的Netty...

    Netty初步学习-HelloNetty.rar

    在"Netty初步学习-HelloNetty.rar"这个压缩包中,我们可以看到包含"第二课netty服务端"的资源,这表明我们将学习如何构建一个简单的 Netty 服务端。下面将详细解释 Netty 的核心概念和步骤,以帮助你入门。 1. **...

    Netty3.x 源码解析

    通过这样的一个例子,我们可以初步感受Netty的使用方法。 源码级别的学习可以帮助开发者从本质上理解Netty的工作原理。为了深入理解Netty,需要关注以下几个关键点: 1. NIO基础:Netty基于Java NIO来实现非阻塞的...

    Java_NIO框架Netty教程.pdf

    了解了上述知识点后,读者应该对Netty的框架有了初步的理解,但Netty的强大远不止这些。在实际应用中,Netty还支持许多高级特性,如SSL/TLS加密通信、零拷贝(zero-copy)和协议栈的灵活定义等。学习和使用Netty,...

    netty-3.2.5终极手册

    - **总结**:通过这些基础示例,读者可以初步了解Netty的工作机制及其核心功能。 #### 四、Netty 架构概览 ##### 1. 富有的缓冲数据结构 Netty内部使用高效的缓冲数据结构(ChannelBuffer),支持快速的数据读写...

    Netty_In_Action第5版电子书及书中源码

    2. **快速入门**:演示如何创建一个简单的服务器和客户端,让读者对Netty的使用有初步认识。 3. **Channel与Pipeline**:讲解Netty的核心组件,包括Channel负责网络通信,Pipeline处理数据传输过程中的事件和处理...

    netty-master.zip

    随附的"netty-课程演示源码(附部分中文注释).zip"和"netty-chapter-1-Netty初步印象之HelloWorld.zip"、"netty-chapter-4-Netty案例雏形代码.zip"提供了一系列的示例代码,帮助读者更好地理解和应用Netty,通过...

    netty-learning:netty-in-action是《 Netty实战》笔记。netty-private-protocol是一个基于Netty私有协议协议实现的

    Netty中大量使用NIO技术,满足高性能、高并发要求。进行初步研究,特作此项目。 源码地址:https://github.com/thinkingfioa/netty-learning 1.为什么要读Netty? 1.1净资产特性 1.2子项目讲解 netty-private-...

    redant:基于Netty一个基于Netty的轻量级Web容器的轻型分布式Web容器

    RedAnt是一个基于Netty的轻量级Web容器 特性: IOC容器:通过@Bean注解可以管理所有对象,通过@Autowired注解进行对象注入 自定义路由:通过@Controller @Mapping @Param注解可以自定义路由 自动参数转换:通过...

    简单了解Java Netty Reactor三种线程模型

    在这种模型中,Acceptor线程池专门处理客户端连接,完成初步的认证等操作,然后将连接注册到后端的子Reactor线程池,由子线程池的线程负责后续的读写操作。这可以进一步提升服务端处理能力,确保连接建立过程的性能...

    ClusterDeviceControlPlatform:Java&Vue.js全栈“放置设备管理云平台『放置部分』』,使用Spring Boot,Netty搭建TCP服务器与上万设备的通信,基于JavaFX的GUI应用程序模拟上万台设备的行为,初步对服务器进行压力测试

    JavaFX投入设备模拟客户端,辅助工具组件“ Python脚本,基于C#和WPF的可视化配置工具” ,简要介绍如下: Java前端服务器:使用Spring Boot 2.0作为基础框架,使用Netty构造TCP服务器与上万台设备组成的通信,采用...

    NettyServer收取数据存库

    在接收到数据时,会触发ChannelInboundHandlerAdapter中的channelRead()或channelReadComplete()方法,我们可以在这里进行数据的初步处理和解析。 描述中提到的"缓存到队列"策略是一种常见的并发编程解决方案,它...

    Rabbit:RPC框架,使用ZooKeeper,Netty,期待良好的设计

    Rabbit从头搭建一个RPC框架,初步打算使用Netty进行数据通信,后期可能会增加BIO和NIO的方式因为是小白入门级别,所以项目代号Rabbit1、首先,先解释下RPC:RPC = Remote Procedure Call ,远程过程调用,它能够通过...

    物联网COAP协议demo-Java版本

    由于物联网中的很多设备都是资源受限型的,即只有少量的内存空间和有限的计算能力,所以传统的HTTP协议应用在物联网上就显得过于庞大而不适用。 IETF的CoRE工作组提出了一种...此应用是java版本的coap协议的基本使用

    CPU 高消耗排查

    #### 二、使用`top`命令进行初步分析 1. **`top`命令基础使用** - `top`是Linux系统中常用的监控工具之一,可以实时查看系统中进程的资源占用情况。 - 命令示例:`[root@3server ~]# top` - 通过`top`命令,我们...

    java_learning_practice:java进阶之路:面试高频算法,akka,多线程,NIO,Netty,SpringBoot,Spark && Flink等

    java_learning_practice学习java的积累示例:面试高频算法,akka,多线程,zookeeper,Disruptor,NIO,Netty,Thrift等项目组成收集总结初步&&大数据方向看到的优秀文章和系列博客;不定期更新pdf阿卡口味akka学习...

    smartacus-mqtt-broker:smartacus-mqtt-broker是基于Java的开源MQTT代理,它完全支持MQTT3.x。使用Netty 4.1.37

    smartacus-mqtt-broker欢迎志同道合的小伙伴一起探讨交流mqtt协议介绍随着工业的快速发展,对设备进行管理和数据采集的需求也越来越高。要解决大量设备的通信...初步协议。压缩消息。可以在不稳定的网络环境中进行良

    异步HttpClient

    目前该组件已经初步测试通过。如果大家也需要这样的组件,可以下载尝试一下。所有关键注释都已经写了,如有不明白可以发送邮件 ath.qu.trues@gmail.com 代码分为3个maven模块。 commons-ext : 实现Promise commons-...

    Java思维导图xmind文件+导出图片

    初步认识RabbitMQ及高可用集群部署 详解RabbitMQ消息分发机制及主题消息分发 RabbitMQ消息路由机制分析 RabbitMQ消息确认机制 Redis redis数据结构分析 Redis主从复制原理及无磁盘复制分析 Redis管道模式...

Global site tag (gtag.js) - Google Analytics