Netty

Netty 下载文件 http://www.open-open.com/lib/view/open1409642102932.html 本实例主要参考的是官网的examples: 点击这里使用场景: 客户端向Netty请求一个文件,Netty服务端下载指定位置文件到客户端。 本实例使用的是Http协议,当然,可以通过简单的修改即可换成TCP协议。 需要注意本实例的关键点是,为了更高效的传输大数据,实例中用到了ChunkedWriteHandler编码器,它提供了以zero-memory-copy方式写文件。 第一步: 先写一个HttpFileServer ? package NettyDemo.file.server; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.stream.ChunkedWriteHandler; /***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\**** * Reserved. BidPlanStructForm.java Created on 2014-8-19 Author: <a * href=mailto:wanghouda@126.com>wanghouda * @Title: HttpFileServer.java * @Package NettyDemo.file.server Description: Version: 1.0 \***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***\***/ public class HttpFileServer { static final int PORT = 8080; public static void main(String[] args) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() {// 有连接到达时会创建一个channel @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new HttpServerCodec()); pipeline.addLast(new HttpObjectAggregator(65536)); pipeline.addLast(new ChunkedWriteHandler()); pipeline.addLast(new FileServerHandler()); } }); Channel ch = b.bind(PORT).sync().channel(); System.err.println("打开浏览器,输入: " + ("http") + "://127.0.0.1:" + PORT + '/'); ch.closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } } 第二步: 再写一个FileServerHandler? ...

netty http client http://www.cnblogs.com/luxiaoxun/p/3959450.html 基于Netty4的HttpServer和HttpClient的简单实现 Netty的主页: http://netty.io/index.html 使用的Netty的版本: netty-4.0.23.Final.tar.bz2 ‐ 15-Aug-2014 (Stable, Recommended) Http 消息格式: Http request: Method path-to-resource HTTPVersion-number Header-name-1: value1 Header-name-2: value2 Optional request body Http response: HTTP/Version-number response-code response-phrase Header-name-1: value1 Header-name-2: value2 Optional response body 实现一个简单的Http请求及响应过程: Client向Server发送http请求。 Server端对http请求进行解析。 Server端向client发送http响应。 Client对http响应进行解析。 Netty中Http request消息格式: Netty中Http response消息格式: 代码实例: Http Server: package com.netty.test; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; 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; import io.netty.handler.codec.http.HttpRequestDecoder; import io.netty.handler.codec.http.HttpResponseEncoder; public class HttpServer { private static Log log = LogFactory.getLog(HttpServer.class); public void start(int port) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { // server端发送的是httpResponse,所以要使用HttpResponseEncoder进行编码 ch.pipeline().addLast(new HttpResponseEncoder()); // server端接收到的是httpRequest,所以要使用HttpRequestDecoder进行解码 ch.pipeline().addLast(new HttpRequestDecoder()); ch.pipeline().addLast(new HttpServerInboundHandler()); } }).option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); ChannelFuture f = b.bind(port).sync(); f.channel().closeFuture().sync(); } finally { workerGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { HttpServer server = new HttpServer(); log.info("Http Server listening on 8844 ..."); server.start(8844); } } 响应请求的HttpServerInboundHandler: ...

netty http文件服务器 http://krisjin.github.io/2015/02/14/netty-file-server/ Http介绍 Http(超文本传输协议)协议是建立在TCP传输协议之上的应用成协议,Http由于便捷、快速的方式,适用于分布式超媒体信息系统。Http是目前Web开发的主流协议,基于Http的应用非常广泛,因此掌握HTTP的开发非常重要。由于netty的HTTP协议栈是基于netty的NIO通信框架开发的,所以netty的HTTP协议也是异步非阻塞的。 具体的关于HTTP及Netyy的实现细节在以后的章节在写,先上实例代码: 实例代码 服务端: public class HttpFileServer { public void run(final int port, final String url) throws InterruptedException { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap boot = new ServerBootstrap(); boot.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast("http-decoder", new HttpRequestDecoder()); ch.pipeline().addLast("http-aggregator", new HttpObjectAggregator(65536)); ch.pipeline().addLast("http-encoder", new HttpResponseEncoder()); ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler()); ch.pipeline().addLast("fileServerHandler", new HttpFileServerHandler(url)); } }); ChannelFuture cf =boot.bind("127.0.0.1",port).sync(); System.out.println("Http文件目录服务器启动: [http://127.0.0.1](http://127.0.0.1):"+port+url); cf.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws InterruptedException { String url="/src/main/java/"; new HttpFileServer().run(8080, url); } } 服务端handler ...

netty 文件上传 http://blog.csdn.net/mcpang/article/details/41140409 【初学与研发之NETTY】netty4之文件上传 标签: netty4文件上传 2014-11-15 04:35 6005人阅读 评论(29) 收藏 举报 分类: JAVA (48) Netty (6) 版权声明: 本文为博主原创文章,未经博主允许不得转载。 客户端: public class UpLoadClient { private StringBuffer resultBuffer = new StringBuffer(); private EventLoopGroup group = null; private HttpDataFactory factory = null; private Object waitObject = new Object(); private ChannelFuture future = null; public UpLoadClient(String host, int port) throws Exception { this.group = new NioEventLoopGroup(); this.factory = new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE); Bootstrap b = new Bootstrap(); b.option(ChannelOption.TCP_NODELAY, true); b.option(ChannelOption.SO_SNDBUF, 1048576*200); b.option(ChannelOption.SO_KEEPALIVE, true); b.group(group).channel(NioSocketChannel.class); b.handler(new UpLoadClientIntializer()); this.future = b.connect(host, port).sync(); } public void uploadFile(String path) { if(path == null) { System.out.println("上传文件的路径不能为null..."); return; } File file = new File(path); if (!file.canRead()) { System.out.println(file.getName() + "不可读..."); return; } if (file.isHidden() || !file.isFile()) { System.out.println(file.getName() + "不存在..."); return; } try { HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, ""); HttpPostRequestEncoder bodyRequestEncoder = new HttpPostRequestEncoder(factory, request, false); bodyRequestEncoder.addBodyAttribute("getform", "POST"); bodyRequestEncoder.addBodyFileUpload("myfile", file, "application/x-zip-compressed", false); List<InterfaceHttpData> bodylist = bodyRequestEncoder.getBodyListAttributes(); if (bodylist == null) { System.out.println("请求体不存在..."); return; } HttpRequest request2 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, file.getName()); HttpPostRequestEncoder bodyRequestEncoder2 = new HttpPostRequestEncoder(factory, request2, true); bodyRequestEncoder2.setBodyHttpDatas(bodylist); bodyRequestEncoder2.finalizeRequest(); Channel channel = this.future.channel(); if(channel.isActive() && channel.isWritable()) { channel.writeAndFlush(request2); if (bodyRequestEncoder2.isChunked()) { channel.writeAndFlush(bodyRequestEncoder2).awaitUninterruptibly(); } bodyRequestEncoder2.cleanFiles(); } channel.closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } } public void shutdownClient() { // 等待数据的传输通道关闭 group.shutdownGracefully(); factory.cleanAllHttpDatas(); } public boolean isCompleted() { while(waitObject != null) { //当通道处于开通和活动时,处于等待 } if(resultBuffer.length() > 0) { if("200".equals(resultBuffer.toString())) { resultBuffer.setLength(0); return true; } } return false; } private class UpLoadClientIntializer extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("decoder", new HttpResponseDecoder()); pipeline.addLast("encoder", new HttpRequestEncoder()); pipeline.addLast("chunkedWriter", new ChunkedWriteHandler()); pipeline.addLast("dispatcher", new UpLoadClientHandler()); } } private class UpLoadClientHandler extends SimpleChannelInboundHandler<HttpObject> { private boolean readingChunks = false; private int succCode = 200; protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception { if (msg instanceof HttpResponse) { HttpResponse response = (HttpResponse) msg; succCode = response.getStatus().code(); if (succCode == 200 && HttpHeaders.isTransferEncodingChunked(response)) { readingChunks = true; } } if (msg instanceof HttpContent) { HttpContent chunk = (HttpContent) msg; System.out.println("【响应】"+succCode+">>"+chunk.content().toString(CharsetUtil.UTF_8)); if (chunk instanceof LastHttpContent) { readingChunks = false; } } if (!readingChunks) { resultBuffer.append(succCode); ctx.channel().close(); } } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { waitObject = null; } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { resultBuffer.setLength(0); resultBuffer.append(500); System.out.println("管道异常: " + cause.getMessage()); cause.printStackTrace(); ctx.channel().close(); } } } 服务端: ...

背景 2017 年前后，我在大连 R 公司负责 R 平台协议解析层的开发和维护。R 平台是一套车联网平台，部署在 H 云上，用来接收和处理车载设备实时上报的数据。 协议解析层有十五个以上的服务，每个服务至少两个节点，负责解析车载设备上报的自定义 TCP 协议，网络通信与连接处理底层由 Netty 实现。平台共有 300 多台虚拟机，单个服务的规格通常不会拉很高，G4 所在的虚拟机配置是 2 CPU、4 GB 内存。 其中 G4 是协议解析层数据量最大的服务，负责解析行车数据，单条消息包含约两百个字段，涵盖车速、转速、门窗锁状态、发动机参数、各类液体温度、油量等实时车况信息。正常 QPS 约 700 条/秒，峰值超过 1000 条/秒，四个节点加起来总 QPS 在 5000 左右。 解析完成后，G4 把数据写入 Redis（实时数据）和 MySQL（行车历史数据，水平分片成 1024 个库），另外还有一些平台内部的 API 调用。MySQL 访问层是在 JDBI 基础上封装了一层。 协议解析层的数据来自上游接入层，接入层向协议解析层推送数据，同时用 RabbitMQ 做消息队列缓冲。 装车增长与早期告警 R 平台接入的设备数量持续增长，每个月有几千台新设备安装到实车上。随着在线行驶车辆增多，G4 的数据量也跟着涨。 大规模故障之前，G4 就开始在早高峰和晚高峰偶发内存告警，高峰过去后十几分钟会自行恢复。当时不清楚具体原因，尝试过把业务线程数从默认值增加到 16，改完之后等服务低峰期重启，观察下一次高峰是否有改善。效果不明确，可能有一定缓解，也可能影响有限，之后就转向其他方向排查，线程数就维持在 16 没有再改。 大规模 OOM 装车量继续增加，某天 G4 出现了大规模 OOM。现象是四个节点滚动离线：一个节点 OOM 重启，剩余节点压力上涨，接着下一个离线，重启后很快再 OOM，如此循环。 节点重启之后撑不了太久，因为上游 RabbitMQ 里已经积压了大量数据，积压时间估计有一两个小时。一个节点恢复，积压数据涌入，Netty 堆外内存里挤满了等待处理的业务数据，内存很快再次撑满。一个节点离线，剩余三个节点分摊的流量更大，内存涨得更快，形成连锁反应。节点之间的 OOM 间隔只有几分钟。 ...

为什么使用堆外内存 Java 堆内存（Heap）受 JVM GC 管理，分配和释放都有 GC 开销，且在 GC 发生时对象地址可能移动（Compacting GC）。进行 I/O 操作时，JVM 必须先把堆内数据拷贝到一块固定地址的堆外缓冲区，再交给内核。 堆外内存（Direct Memory / Off-Heap）直接由操作系统管理，有以下优势： 零拷贝：内核可以直接读写堆外内存，省掉堆内 → 堆外的一次拷贝 不受 GC 暂停影响：地址固定，内核持有指针不会因 GC 而失效 更可预测的延迟：不参与 Young/Old GC，减少 STW（Stop-The-World）概率 代价是分配/释放成本更高，且不会被 GC 自动回收，需要手动管理生命周期。 ByteBuf：Netty 的核心缓冲区抽象 Netty 用 ByteBuf 替代 JDK 的 ByteBuffer，消除了 flip()/rewind() 的使用负担，并同时支持堆内和堆外两种底层存储。 分类 ByteBuf ├── HeapByteBuf 堆内，byte[] 存储，GC 管理 ├── DirectByteBuf 堆外，java.nio.DirectByteBuffer 存储 └── CompositeByteBuf 零拷贝组合多个 ByteBuf（可混合堆内/堆外） 每种又分 Pooled（池化）和 Unpooled（非池化）两个版本： 类型 分配方式 适用场景 PooledDirectByteBuf 从内存池取 高频、短生命周期 I/O UnpooledDirectByteBuf 每次调用 malloc 低频、大块、长期持有 PooledHeapByteBuf 从内存池取 不需要 native I/O 的场景 UnpooledHeapByteBuf 每次 new byte[] 简单测试/工具代码 零拷贝（Zero-copy） OS 层面的 Zero-copy 指避免在用户态（User-space）与内核态（Kernel-space）之间拷贝数据，例如 Linux 的 mmap、sendfile。Netty 的 Zero-copy 完全在用户态（Java 层面），核心目标是减少 ByteBuf 之间不必要的内存拷贝。 ...

Reactor / Dispatcher 模式 了解 Reactor 模式，就要先从事件驱动的开发方式说起。 我们知道，服务器开发，CPU 的处理速度远高于 IO 速度，为了避免 CPU 因为 IO 而阻塞，好一点的方法是多进程或线程处理，但这会带来一些进程切换的开销。 这时先驱者找到了事件驱动，或者叫回调的方法。这种方式就是，应用向一个中间人注册一个回调 (Event handler)，当 IO 就绪后，这个中间人产生一个事件，并通知此 handler 进行处理。这种回调的方式，也实现了"好莱坞原则" - “Don’t call us, we’ll call you.” 那在 IO 就绪这个事件后，谁来充当这个中间人？Reactor 模式的答案是: 有一个不断等待和循环的单独进程 (线程) 来做这件事，它接受所有 handler 的注册，并负责向操作系统查询 IO 是否就绪，在就绪后用指定的 handler 进行处理，这个角色的名称就叫做 Reactor。 事件驱动, 回调函数, reactor, 响应式编程 事件驱动是概念，回调函数是实现方式。不用回调函数，也可以实现事件驱动。例如：把事件消息发送到队列，另外一个进程取队列处理即可 (没有回调函数）。事件驱动的本质特征：中心轮询机制。event loop的loop是轮询。轮询的目的是什么？感知！对象发生变化，如何感知这种变化？不断的循环查询，loop探测！系统n个对象，每个对象一个for循环 探测彼此的变化？nonono……建立一个轮询中心，这个轮询中心去轮询每个对象，这就是事件驱动。发生了变化，通知感兴趣的对象，怎么处理？就是定义一个回调函数。事件驱动，属于“感知层”的概念；轮询中心，往往就是操作系统本身；对于浏览器而言，就是浏览器本身。也就是系统是轮询中心，你定义 函数，系统调用你定义的函数。对比：系统定义api，你调用api。谁定义函数，谁调用，角色颠倒了！api：系统定义的函数，你去调用； 事件驱动：你定义的回调函数，被系统调用。还是没有懂？事件驱动，就是“哨兵模式”！哨兵轮询环境信息，你就安心睡大觉好了，不用每个人都轮询环境。发生了事件，哨兵 (操作系统/浏览器/轮询中心）负责通知你！怎么处理这个消息，是你的责任！ 作者：fdego 链接：https://www.zhihu.com/question/30396023/answer/447966119 来源：知乎 著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。 事件驱动, reactor The reactor design pattern is an event handling pattern for handling service requests delivered concurrently to a service handler by one or more inputs. https://en.wikipedia.org/wiki/Reactor_pattern ...