当前位置： 首页 > article >正文

Netty从入门到进阶（二）

article 2025/6/12 8:42:58

二、Netty入门

1. 概述

1.1 Netty是什么

Netty is an asynchronous event-driven network application framework for rapid  development of maintainable high performance protocol servers & clients.

Netty是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端。

1.2 Netty的作者

他还是另一个著名网络应用框架Mina的重要贡献者。

1.3 Netty的地位

Netty在Java网络应用框架中的地位就好比：Spring框架在JavaEE开发中的地位

以下框架都使用了Netty，因为它们有网络通信需求：

  • Cassandra - nosql 数据库

  • Spark - 大数据分布式计算框架

  • Hadoop - 大数据分布式存储框架

  • RocketMQ - ali 开源的消息队列

  • ElasticSearch - 搜索引擎

  • gRPC - rpc 框架

  • Dubbo - rpc 框架

  • Spring 5.x - flux api 完全抛弃了 tomcat ，使用 netty 作为服务器端

  • Zookeeper - 分布式协调框架

1.4 Netty的优势

Netty vs NIO，工作量大，bug多

  • NIO 需要自己构建协议
  • Netty 解决了TCP传输问题，如粘包、半包
  • NIO epoll空轮询导致 CPU 100%
  • Netty 对API进行增强，使之更易用，如FastThreadLocal => ThreadLocal, ByteBuf => ByteBuffer

Netty vs 其它网络应用框架

  • Mina由apache维护，将来3.x版本可能会有较大重构，破坏API向下兼容性，而Netty的开发迭代更迅速，API更简洁、文档更优秀

久经考验，16年，Netty版本

  • 2.x 2004
  • 3.x 2008
  • 4.x 2013
  • 5.x 已废弃（没有明显的性能提升，维护成本高）

2. Hello World

2.1 目标

开发一个简单的服务器端和客户端

  • 客户端向服务器端发送hello, world
  • 服务器仅接收，不返回

2.2 服务器端

package cn.itcast.netty.c2;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LoggingHandler;public class HelloServer {public static void main(String[] args) {// 1. 启动器，负责组装 netty 组件，启动服务器new ServerBootstrap()// 2. BossEventLoop, WorkerEventLoop(selector,thread), group 组.group(new NioEventLoopGroup())// 3. 选择 服务器的 ServerSocketChannel 实现.channel(NioServerSocketChannel.class) // OIO BIO// boss 负责处理连接, worker(child) 负责处理读写，决定了 worker(child) 能执行哪些操作（handler）.childHandler(//  channel 代表和客户端进行数据读写的通道 Initializer 初始化，负责添加别的 handlernew ChannelInitializer<NioSocketChannel>() {// 连接建立后，调用初始化方法@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {// 添加具体 handlerch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringDecoder()); // 5. 将 ByteBuf 转换为字符串ch.pipeline().addLast(new ChannelInboundHandlerAdapter() { // 6. 自定义 handler@Override // 读事件public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println(msg); // 打印上一步转换好的字符串}});}})// 4. 绑定监听端口.bind(8080);}
}

代码解读

  • 1处，创建NioEventLoopGroup，可以简单理解为线程池 + selector
  • 2处，选择服务Socket实现类，其中有NioServerSocketChannel表示基于NIO的服务器端实现，其它实现还有

  • 3处，为啥方法叫childHandler，是接下来添加的处理器都是给SocketChannel用的，而不是给ServerSocketChannel。ChannelInitializer处理器（仅执行一次），它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
  • 4处，ServerSocketChannel绑定的监听端口
  • 5处，SocketChannel的处理器，解码 ByteBuf => String
  • 6处，SocketChannel的业务处理器，使用上一个处理器的处理结果

输出

15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] REGISTERED
15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] ACTIVE
15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] READ: 12B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f 2c 20 77 6f 72 6c 64             |hello, world    |
+--------+-------------------------------------------------+----------------+
hello, world
15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] READ COMPLETE

2.3 客户端

package cn.itcast.netty.c2;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;import java.net.InetSocketAddress;public class HelloClient {public static void main(String[] args) throws InterruptedException {// 启动类new Bootstrap()// 1. 添加 EventLoop.group(new NioEventLoopGroup())// 2. 选择客户端 channel 实现.channel(NioSocketChannel.class)// 3. 添加处理器.handler(new ChannelInitializer<NioSocketChannel>() {@Override // 在连接建立后被调用protected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());  // 8. 将 字符串 转换为 ByteBuf}})// 4. 连接到服务器.connect(new InetSocketAddress("localhost", 8080)).sync()  // 5. 阻塞方法，直到连接建立.channel()  // 6. 代表客户端与服务器端的连接对象// 7. 向服务器发送数据.writeAndFlush("hello, world");  // 然后调用StringEncoder()方法，把字符串转为ByteBuf}
}

代码解读

  • 1处，创建NioEventLoopGroup，同Server
  • 2处，选择客户Socket实现类，NioSocketChannel表示基于NIO的客户端实现，其它还有

  • 3处，添加SocketChannel的处理器，ChannelInitializer处理器（仅执行一次），它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
  • 4处，指定要连接的服务器和端口
  • 5处，Netty中很多方法都是异步的，如connect，这时需要使用sync方法等待connect建立连接完毕
  • 6处，获取channel对象，它即为通道抽象，可以进行数据读写操作
  • 7处，写入消息并清空缓冲区
  • 8处，消息会经过通道handler处理，这里是将 String => ByteBuf 发出
  • 数据经过网络传输，到达服务器端，服务器端5和6处的handler都被触发，走完一个流程

2.4 流程梳理

💡提示

一开始需要树立正确的观念

  • 把channel理解为数据的通道
  • 把msg理解为流动的数据，最开始输入是ByteBuf，但经过pipeline的加工，会变成其它类型对象，最后输出又变成ByteBuf
  • 把handler理解为数据的处理工序
    • 工序有多道，合在一起就是pipeline，pipeline负责发布事件（读、读取完成...）传播给每个handler，handler对自己感兴趣的事件进行处理（重写了相应事件处理方法）
    • handler份Inbound（入站） 和 Outbound（出站）两类
  • 把eventLoop理解为处理数据的工人
    • 工人可以管理多个channel的io操作，并且一旦工人负责了某个channel，就要负责到底（绑定）
    • 工人既可以执行io操作，也可以进行任务处理，每位工人有任务队列，队列力可以堆放多个channel的待处理任务，任务分为普通任务、定时任务
    • 工人按照pipeline顺序，依次按照handler的规划（代码）处理数据，可以为每道工序指定不同的工人。

3. 组件

3.1 EventLoop

EventLoop 本质是一个单线程执行器（同时维护了一个Selector），里面有run方法处理Channel上源源不断的IO事件。

它的继承关系比较复杂：

  • 一条线是继承自 j.u.c.ScheduledExecutorService，因此包含了线程池中的所有方法；
  • 另一条线是继承自netty自己的OrderedEventExecutor
    • 提供了boolean inEventLoop(Thread thread)方法，判断一个线程是否属于此EventLoop
    • 提供了parent方法来看看自己属于哪个EventLoopGroup

EventLoopGroup是一组EventLoop，Channel一般会调用EventLoopGroup的register方法来绑定其中一个EventLoop，后续这个Channel上的IO事件都由此EventLoop来处理（保证了IO事件处理时的线程安全）

  • 继承自netty自己的EventExecutorGroup
    • 实现了iterable接口提供遍历EventLoop的能力
    • 另有next方法获取集合中下一个EventLoop

示例代码：

package cn.itcast.netty.c2;import io.netty.channel.DefaultEventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.TimeUnit;@Slf4j
public class TestEventLoop {public static void main(String[] args) {// 1. 创建事件循环组EventLoopGroup group = new NioEventLoopGroup(2);  // IO事件、普通任务、定时任务
//        EventLoopGroup group2 = new DefaultEventLoop();  // 普通任务、定时任务// 2. 获取下一个事件循环对象System.out.println(group.next());System.out.println(group.next());System.out.println(group.next());// 3. 执行普通任务（可以用来执行耗时较长的任务）group.next().submit(() -> {try {Thread.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}log.debug("ok");});// 4. 执行定时任务group.next().scheduleAtFixedRate(() -> {log.debug("定时任务!");}, 0, 1, TimeUnit.SECONDS);log.debug("main");}
}

输出

io.netty.channel.nio.NioEventLoop@553f17c
io.netty.channel.nio.NioEventLoop@4f7d0008
io.netty.channel.nio.NioEventLoop@553f17c
17:12:20 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!
17:12:20 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestEventLoop - ok
17:12:20 [DEBUG] [main] c.i.n.c.TestEventLoop - main
17:12:21 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!
17:12:22 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!
17:12:23 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!

💡优雅关闭

优雅关闭shutdownGracefully方法。该方法会首先切换EventLoopGroup到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出。

演示NioEventLoop处理IO事件

服务器端两个 nio worker工人：

package cn.itcast.netty.c2;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.nio.charset.Charset;@Slf4j
public class EventLoopServer {public static void main(String[] args) {new ServerBootstrap()// boss 和 worker// boss只负责SeverSocketChannel上的accept事件，worker只负责socketChannel上的读写操作.group(new NioEventLoopGroup(), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;log.debug(buf.toString(Charset.defaultCharset()));}});}}).bind(8080);}
}

客户端：

package cn.itcast.netty.c2;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;import java.net.InetSocketAddress;public class EventLoopClient {public static void main(String[] args) throws InterruptedException {// 1. 启动类Channel channel = new Bootstrap()// 2. 添加 EventLoop.group(new NioEventLoopGroup())// 3. 选择客户端 channel 实现.channel(NioSocketChannel.class)// 4. 添加处理器.handler(new ChannelInitializer<NioSocketChannel>() {@Override // 在连接建立后被调用protected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());  // 将 字符串 转换为 ByteBuf}})// 5. 连接到服务器.connect(new InetSocketAddress("localhost", 8080)).sync()  // 阻塞方法，直到连接建立.channel();  // 代表客户端与服务器端的连接对象}
}

两个工人轮流处理channel，但工人与channel之间进行了绑定

增加两个非 nio 工人：

package cn.itcast.netty.c2;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.nio.charset.Charset;@Slf4j
public class EventLoopServer {public static void main(String[] args) {// 细分2：创建一个独立的EventLoopGroup（增加两个非nio工人）EventLoopGroup group = new DefaultEventLoop();new ServerBootstrap()// boss 和 worker// 细分1：boss只负责SeverSocketChannel上的accept事件，worker只负责socketChannel上的读写操作.group(new NioEventLoopGroup(), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast("handler1", new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;log.debug(buf.toString(Charset.defaultCharset()));ctx.fireChannelRead(msg);  // 让消息传递给下一个handler}}).addLast(group, "handler2", new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;log.debug(buf.toString(Charset.defaultCharset()));}});}}).bind(8080);}
}

可以看到nio工人和非nio工人也分别绑定了channel，而我们自己的handler由非nio工人执行。

💡handler执行中如何换人？

关键代码：io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);// 下一个 handler 的事件循环是否与当前的事件循环是同一个线程EventExecutor executor = next.executor();  // 返回下一个handler的eventLoop// 是，直接调用if (executor.inEventLoop()) {  // 当前handler中的线程，是否和eventLoop是同一个线程next.invokeChannelRead(m);  // 直接调用} // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）else {executor.execute(new Runnable() {  // 下一个handler线程@Overridepublic void run() {next.invokeChannelRead(m);}});}// 如果两个handler绑定的是同一个线程，那么就直接调用； 否则，把要调用的代码封装为一个任务对象，由下一个handler的线程来调用
}
  • 如果两个handler绑定的是同一个线程，那么就直接调用；
  • 否则，把要调用的代码封装为一个任务对象，由下一个handler的线程来调用

3.2 Channel

channel的主要作用：

  • close()可以用来关闭channel
  • closeFuture()用来处理channel的关闭
    • sync方法作用是同步等待channel关闭
    • 而addListener方法是异步等待channel关闭
  • pipeline()方法添加处理器
  • write()方法将数据写入（要调用flush方法才会从缓冲区发出）
  • writeAndFlush()方法将数据写入并刷出（立刻发出）

ChannelFuture

这是刚才的客户端代码：

new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect("127.0.0.1", 8080).sync().channel().writeAndFlush(new Date() + ": hello world!");

现在把他拆开来看

ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect("127.0.0.1", 8080); // 1channelFuture.sync().channel().writeAndFlush(new Date() + ": hello world!");
  • 1处返回的是ChannelFuture对象，它的作用是利用channel()方法来获取Channel对象

注意：connect方法是异步的，意味着不等连接建立，方法执行就返回了。因此ChannelFuture对象中不能【立刻】获得到正确的Channel对象。

实验如下：

ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect("127.0.0.1", 8080);System.out.println(channelFuture.channel()); // 1
channelFuture.sync(); // 2
System.out.println(channelFuture.channel()); // 3
  • 执行到1时，连接未建立，打印：[id: 0x2e1884dd]
  • 执行到2时，sync方法是同步等待连接建立完成
  • 执行到3时，连接建立了，打印：[id: 0x2e1884dd, L:/127.0.0.1:57191 - R:/127.0.0.1:8080]

除了用sync方法可以让异步操作同步以外，还可以使用回调的方式：

ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect("127.0.0.1", 8080);
System.out.println(channelFuture.channel()); // 1
channelFuture.addListener((ChannelFutureListener) future -> {System.out.println(future.channel()); // 2
});
  • 执行到1时，连接未建立，打印：[id: 0x749124ba]
  • ChannelFutureListener会在连接建立时被调用（其中operationComplete方法），因此执行到2时，连接肯定建立了，打印：[id: 0x749124ba, L:/127.0.0.1:57351 - R:/127.0.0.1:8080]
package cn.itcast.netty.c2;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.Date;@Slf4j
public class EventLoopClient {public static void main(String[] args) throws InterruptedException {// 带有Future，Promise的类型都是和异步方法配套使用，用来处理结果ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());  // 将 字符串 转换为 ByteBuf}})// 1. 连接到服务器// 异步非阻塞, main发起了调用，真正执行connect连接的是 nio 线程.connect(new InetSocketAddress("localhost", 8080));// 2.1 使用sync方法同步处理结果/*channelFuture.sync()  // 阻塞当前线程，直到nio线程连接建立完毕.channel().writeAndFlush(new Date() + ": hello world!");*/// 2.2 使用addListener(回调对象)方法异步处理结果channelFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {// 在nio线程连接建立好之后，会调用operationCompleteChannel channel = future.channel();log.debug("{}", channel);channel.writeAndFlush(new Date() + "hello, world!");}});}
}

CloseFuture

package cn.itcast.netty.c2;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.Scanner;@Slf4j
public class CloseFutureClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();ChannelFuture channelFuture = new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress("localhost", 8080));Channel channel = channelFuture.sync().channel();log.debug("{}", channel);new Thread(() -> {Scanner scanner = new Scanner(System.in);while (true) {String line = scanner.nextLine();if("q".equals(line)) {channel.close();  // 异步操作
//                    log.debug("处理关闭之后的操作");  // 不能在这里善后break;}channel.writeAndFlush(line);}}, "input").start();// 获取CloseFuture对象， 1）同步处理关闭 2）异步处理关闭ChannelFuture closeFuture = channel.closeFuture();// 1）同步处理关闭/*log.debug("waiting close...");closeFuture.sync();  // 阻塞log.debug("处理关闭之后的操作");*/// 2) 异步处理关闭closeFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture channelFuture) throws Exception {log.debug("处理关闭之后的操作");  // nio线程调用group.shutdownGracefully();  // 优雅停止}});}
}

💡异步提升的是什么

  • 有些同学看到这里会有疑问：为什么不在一个线程中去执行建立连接、去执行关闭 channel，那样不是也可以吗？非要用这么复杂的异步方式：比如一个线程发起建立连接，另一个线程去真正建立连接

  • 还有同学会笼统地回答，因为 netty 异步方式用了多线程、多线程就效率高。其实这些认识都比较片面，多线程和异步所提升的效率并不是所认为的

思考下面的场景：4个医生给人看病，每个病人花费20分钟，而且医生看病的过程中是以病人为单位的，一个病人看完了，才能看下一个病人。假设病人源源不断地来，可以计算一下4个医生一天工作8小时，处理的病人总数是 4 * 8 * 3  = 96。

经研究发现，看病可以细分为四个步骤，经拆分后每个步骤需要5分钟，如下：

因此，可以做如下优化，只有一开始，医生2、3、4分别需要等待5、10、15分钟才能执行工作，但只要后续病人源源不断地来，他们就能够满负荷工作，并且处理病人的能力提高到了4 * 8 * 12（20分钟一个 -> 5分钟一个），效率几乎是原来的4倍。

要点

  • 单线程没法异步提高效率，必须配合多线程、多核CPU才能发挥异步的优势；
  • 异步并没有压缩响应时间，反而有所增加；
  • 合理进行任务拆分，也是利用异步的关键。

3.3 Future & Promise

在异步处理时，经常用到这两个接口。首先要说明netty中的Future与jdk中的Future同名，但是是两个接口，netty的Future继承自jdk的Future，而Promise又对netty Future进行了扩展：

  • jdk Future只能同步等待任务结束（或成功、或失败）才能得到结果
  • netty Future可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等待任务结束；
  • netty Promise不仅有netty Future的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器。
功能/名称jdk Futurenetty FuturePromise
cancel取消任务
isCanceled任务是否取消
isDone任务是否完成，不能区分成功还是失败
get获取任务结果，阻塞等待
getNow获取任务结果，非阻塞，还未产生结果时返回null

await等待任务结束，如果任务失败，不会抛异常，而是通过isSuccess判断
sync等待任务结束，如果任务失败，抛出异常
isSuccess判断任务是否成功
cause获取失败信息，非阻塞；如果没有失败，返回null
addListener添加回调，异步接收结果
setSuccess设置成功结果
setFailure设置失败结果

例1：jdk中Future

package cn.itcast.netty.c2;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.*;@Slf4j
public class TestJdkFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 线程池ExecutorService service = Executors.newFixedThreadPool(2);// 2. 提交任务Future<Integer> future = service.submit(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {log.debug("执行计算...");Thread.sleep(1000);return 50;}});// 3. 主线程通过future获取结果log.debug("等待结果...");log.debug("结果是: {}", future.get());  // 阻塞等待}
}

输出

09:28:35 [DEBUG] [main] c.i.n.c.TestJdkFuture - 等待结果...
09:28:35 [DEBUG] [pool-1-thread-1] c.i.n.c.TestJdkFuture - 执行计算...
09:28:36 [DEBUG] [main] c.i.n.c.TestJdkFuture - 结果是: 50

例2：Netty中的Future

package cn.itcast.netty.c2;import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.ExecutionException;@Slf4j
public class TestNettyFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();EventLoop eventLoop = group.next();Future<Integer> future = eventLoop.submit(() -> {log.debug("执行计算...");Thread.sleep(1000);return 70;});log.debug("等待结果...");log.debug("getNow(), 结果是: {}", future.getNow());  // 非阻塞log.debug("结果是: {}", future.get());  // 阻塞future.addListener(new GenericFutureListener<Future<? super Integer>>() {@Overridepublic void operationComplete(Future<? super Integer> future) throws Exception {log.debug("接收结果: {}", future.getNow());}});}
}

输出

09:35:47 [DEBUG] [main] c.i.n.c.TestNettyFuture - 等待结果...
09:35:47 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestNettyFuture - 执行计算...
09:35:47 [DEBUG] [main] c.i.n.c.TestNettyFuture - getNow(), 结果是: null
09:35:48 [DEBUG] [main] c.i.n.c.TestNettyFuture - 结果是: 70
09:35:48 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestNettyFuture - 接收结果: 70

例3：Netty中的Promise

package cn.itcast.netty.c2;import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.ExecutionException;@Slf4j
public class TestNettyPromise {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 创建EventLoop对象EventLoop eventLoop = new NioEventLoopGroup().next();// 2. 主动创建Promise，结果容器DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);new Thread(() -> {// 3. 任意一个线程执行计算，计算完毕后向promise填充结果log.debug("开始计算...");try {int i = 1 / 0;Thread.sleep(1000);promise.setSuccess(80);} catch (InterruptedException e) {e.printStackTrace();promise.setFailure(e);}}).start();// 4. 接收结果log.debug("等待结果...");log.debug("结果是: {}", promise.get());}
}

输出

09:43:06 [DEBUG] [Thread-0] c.i.n.c.TestNettyPromise - 开始计算...
09:43:06 [DEBUG] [main] c.i.n.c.TestNettyPromise - 等待结果...
Exception in thread "Thread-0" java.lang.ArithmeticException: / by zero

3.4 Handler & Pipeline

ChannelHandler用来处理Channel上的各种事件，分为入站、出站两种。所有ChannelHnadler被连成一串，就是Pipeline

  • 入站处理器通常是ChannelInboundHandlerAdapter的子类，主要用来读取客户端数据，写回结果；
  • 出站处理器通常是ChannelOutboundHandlerAdapter的子类，主要对写回结果进行加工

打个比喻，每个Channel是一个产品的加工车间，Pipeline是车间中的流水线，ChannelHandler就是流水线上的各道工序，而ByteBuf就是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品。

服务器端：

package cn.itcast.netty.c2;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;@Slf4j
public class TestPipeline {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {// 1. 通过channel拿到pipelineChannelPipeline pipeline = ch.pipeline();// 2. 添加处理器 head ⇄ h1 ⇄ h2 ⇄ h3 ⇄ h4 ⇄ h5 ⇄ h6 ⇄ tailpipeline.addLast("h1", new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug("1");super.channelRead(ctx, msg);  // 1// ctx.fireChannelRead(msg);}});pipeline.addLast("h2", new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug("2");super.channelRead(ctx, msg);  // 2}});pipeline.addLast("h3", new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug("3");super.channelRead(ctx, msg);  ch.writeAndFlush(ctx.alloc().buffer().writeBytes("server...".getBytes()));  // 3}});pipeline.addLast("h4", new ChannelOutboundHandlerAdapter() {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug("4");super.write(ctx, msg, promise);  // 4}});pipeline.addLast("h5", new ChannelOutboundHandlerAdapter() {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug("5");super.write(ctx, msg, promise);  // 5}});pipeline.addLast("h6", new ChannelOutboundHandlerAdapter() {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug("6");super.write(ctx, msg, promise);  // 6}});}}).bind(8080);}
}

客户端：

package cn.itcast.netty.c2;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.util.Scanner;@Slf4j
public class CloseFutureClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();ChannelFuture channelFuture = new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress("localhost", 8080));Channel channel = channelFuture.sync().channel();log.debug("{}", channel);new Thread(() -> {Scanner scanner = new Scanner(System.in);while (true) {String line = scanner.nextLine();if("q".equals(line)) {channel.close();  // 异步操作
//                    log.debug("处理关闭之后的操作");  // 不能在这里善后break;}channel.writeAndFlush(line);}}, "input").start();// 获取CloseFuture对象， 1）同步处理关闭 2）异步处理关闭ChannelFuture closeFuture = channel.closeFuture();// 1）同步处理关闭/*log.debug("waiting close...");closeFuture.sync();  // 阻塞log.debug("处理关闭之后的操作");*/// 2) 异步处理关闭closeFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture channelFuture) throws Exception {log.debug("处理关闭之后的操作");  // nio线程调用group.shutdownGracefully();  // 优雅停止}});}
}

客户端输出：

aaa
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61                                        |aaa             |
+--------+-------------------------------------------------+----------------+
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] FLUSH
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] READ: 9B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 73 65 72 76 65 72 2e 2e 2e                      |server...       |
+--------+-------------------------------------------------+----------------+
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] READ COMPLETE

服务器端输出：先进后出

10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 1
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 2
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 3
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 6
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 5
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 4

结论：可以看到ChannelInboundHandlerAdapter是按照addLast的顺序执行的，而ChannelOutboundHandlerAdapter是按照addLast的逆序执行的。ChannelPipeline的实现是一个ChannelHandlerContext（包装了ChannelHandler）组成的双向链表。

  • 入站处理器中，ctx.fireChannelRead(msg)是调用下一个入站处理器
    • 如果注释掉1处代码，则仅会打印 1
    • 如果注释掉2处代码，则仅会打印 1 2
  • 3处的ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("server...".getBytes()))会从尾部开始触发后续出站处理器的执行
    • 如果注释掉3处代码，则仅会打印 1 2 3
  • 类似的，出站处理器中，super.write(ctx, msg, promise)的调用会触发上一个出站处理器
    • 如果注释掉6处代码，则仅会打印 1 2 3 6
  • ctx.channel().write(msg) vs ctx.write(msg)
    • 都是触发出站处理器的执行
    • ctx.channel().write(msg)从尾部开始查找出站处理器
    • ctx.write(msg)是从当前节点找上一个出站处理器
    • 3处的cxt.channel().write(msg)如果改为ctx.write(msg)，则仅会打印1 2 3，因为节点3之前没有其它出站处理器了
    • 6处的ctx.write(msg, promise)如果改为ctx.channel().write(msg)会打印1 2 3 6 6 6 ...，因为ctx.channel().write()是从尾部tail开始查找，结果又是节点6自己

借助EmbeddedChannel进行测试

package cn.itcast.netty.c2;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.embedded.EmbeddedChannel;
import lombok.extern.slf4j.Slf4j;@Slf4j
public class TestEmbeddedChannel {public static void main(String[] args) {ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug("1");super.channelRead(ctx, msg);}};ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug("2");super.channelRead(ctx, msg);}};ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter() {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug("3");super.write(ctx, msg, promise);}};ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter() {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {log.debug("4");super.write(ctx, msg, promise);}};EmbeddedChannel channel = new EmbeddedChannel(h1, h2, h3, h4);// 模拟入站操作channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes()));  // 1 2// 模拟出站操作channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("world".getBytes()));  // 4 3}
}

输出

10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 1
10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 2
10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 4
10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 3

3.5 ByteBuf

ByteBuf是对字节数据的封装

1）创建

package cn.itcast.netty.c2;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
import static io.netty.util.internal.StringUtil.NEWLINE;public class TestByteBuf {public static void main(String[] args) {ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();  // 可以动态扩容 基于直接内存的ByteBuf
//        System.out.println(buf);  // PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 256)log(buf);StringBuilder sb = new StringBuilder();for (int i = 0; i < 300; i++) {sb.append("a");}buf.writeBytes(sb.toString().getBytes());
//        System.out.println(buf);  // PooledUnsafeDirectByteBuf(ridx: 0, widx: 300, cap: 512)log(buf);}private static void log(ByteBuf buffer) {int length = buffer.readableBytes();int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;StringBuilder buf = new StringBuilder(rows * 80 * 2).append("read index:").append(buffer.readerIndex()).append(" write index:").append(buffer.writerIndex()).append(" capacity:").append(buffer.capacity()).append(NEWLINE);appendPrettyHexDump(buf, buffer);System.out.println(buf.toString());}
}

输出

read index:0 write index:0 capacity:256

read index:0 write index:300 capacity:512
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000010| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000020| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000030| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000040| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000050| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000060| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000070| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000080| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000090| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000a0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000b0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000c0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000d0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000e0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000f0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000100| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000110| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000120| 61 61 61 61 61 61 61 61 61 61 61 61             |aaaaaaaaaaaa    |
+--------+-------------------------------------------------+----------------+

2）直接内存 vs 堆内存

堆内存的分配效率较高，但读写效率较低，且受到垃圾回收影响；而直接内存分配效率较低，读写效率较高（零拷贝）。

可以使用下面的代码来创建池化基于堆的ByteBuf：

ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

也可以使用下面的代码来创建池化基于直接内存的bytebuf：

ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);
  • 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起使用
  • 直接内存堆GC压力小，因为这部分内存不受JVM垃圾回收的管理，但也要注意及时主动释放。

3）池化 vs 非池化

池化的最大意义在于可以重用ByteBuf，优点有：

  • 没有池化，则每次都得创建新的ByteBuf实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加GC压力；
  • 有了池化，则可以重用池中ByteBuf实例，并且采用了与jemalloc类似的内存分配算法提升分配效率；
  • 高并发时，池化功能更节约内存，减少内存溢出的可能。

池化功能是否开启，可以通过下面的系统环境变量或虚拟机参数来设置：

-Dio.netty.allocator.type={unpooled|pooled}
  • 4.1以后，非Android平台默认启动池化实现，Android平台启用非池化实现
  • 4.1之前，池化功能还不成熟，默认是非池化实现。

4）组成

ByteBuf由四部分组成：

  • 最开始读写指针都在0位置；
  • 废弃字节是指已经读取过的部分；
  • 最大容量为Integet.MAX_VALUE（即2^31 -1，约2GB)，底层限制：
    • 堆内存：受JVM堆大小限制（可通过-Xmx调整）；
    • 直接内存：受系统内存和-XX:MaxDirectMemorySize限制

5）写入

方法签名含义备注
writeBoolean(boolean value)写入boolean值用一字节 01 | 00 代表 true | false
writeByte(int value)写入byte值
writeShort(int value)写入short值
writeInt(int value)写入int值Big Endian（大端），即0x250，写入后 00 00 02 50
writeIntLE(int value)写入int值Little Endian（小端），即0x250，写入后 50 02 00 00
writeLong(long value)写入long值
writeChar(int value)写入char值
writeFloat(float value)写入float值
writeDouble(double value)写入double值
writeBytes(ByteBuf src)写入netty的ByteBuf
writeBytes(byte[] src)写入byte[]
writeBytes(ByteBuffer src)写入nio的ByteBuffer
int writeCharSequence(CharSequence sequence, Charset charset)写入字符串

注意：

  • 这些方法未指明返回值的，其返回值都是ByteBuf，意味着可以链式调用
  • 网络传输，默认习概是Big Endian

先写入4个字节：

buffer.writeBytes(new byte[]{1, 2, 3, 4});
log(buffer);

结果是：

read index:0 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04                                     |....            |
+--------+-------------------------------------------------+----------------+

再写入一个int整数，也是4个字节

buffer.writeInt(5);
log(buffer);

结果是：

read index:0 write index:8 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05                         |........        |
+--------+-------------------------------------------------+----------------+

还有一类方法是set开头的一系列方法，也可以写入数据，但不会改变写指针位置。

6）扩容

这时再写入一个int整数时，容量不够了（初始容量是10），这时会扩容

buffer.writeInt(6);
log(buffer);

扩容规则是：

  • 如果写入后数据大小未超过512，则选择下一个16的整数倍，例如写入后大小为12，则扩容后capacity是16；
  • 如果写入后数据大小超过512，则选择下一个2^n，例如写入后大小为513，则扩容后capacity是2^10=1024（z^9=512已经不够了）；
  • 扩容不能超过max capacity，否则会报错

结果是：

read index:0 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05 00 00 00 06             |............    |
+--------+-------------------------------------------------+----------------+

7）读取

例如读了4次，每次一个字节

System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
log(buffer);

读过的内容，就属于废弃部分了，再读只能读那些尚未读取的部分：

1
2
3
4
read index:4 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06                         |........        |
+--------+-------------------------------------------------+----------------+

如果需要重复读取int整数5，怎么办？

  • 可以在read前先做个标记mark
buffer.markReaderIndex();
System.out.println(buffer.readInt());
log(buffer);

结果：

5
read index:8 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 06                                     |....            |
+--------+-------------------------------------------------+----------------+

这时要重复读取的话，重置到标记位置reset：

buffer.resetReaderIndex();
log(buffer);

这时：

read index:4 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06                         |........        |
+--------+-------------------------------------------------+----------------+

还有一种办法就是采用get开头的一系列方法，这些方法不会改变read index。

8）retain & release

由于Netty中有堆外内存的ByteBuf实现，堆外内存最好还是手动来释放，而不是等GC垃圾回收。

  • UnpooledHeapByteBuf使用的是JVM内存，只需等GC回收内存即可；
  • UnpooledDirectByteBuf使用的就是直接内存了，需要特殊的方法来回收内存；
  • PooledByteBuf和它的子类使用了池化机制，需要更复杂的规则来回收内存

Netty采用了引用计数法来控制回收内存，每个ByteBuf都实现了ReferenceCounted接口

  • 每个ByteBuf对象的初始计数为1；
  • 调用release方法计数减1，如果计数为0，ByteBuf内存被回收
  • 调用retain方法计数加1，表示调用者没用完之前，其它handler即使调用了release也不会造成内存回收；
  • 当计数为0时，底层内存会被回收，这时即使ByteBuf对象还在，其各个方法均无法正常使用。

问题：谁来负责release呢？

不是我们想象的（一般情况下）

ByteBuf buf = ...
try {...
} finally {buf.release();
}

请思考：因为pipeline的存在，一般需要将ByteBuf传递给下一个ChannelHandler，如果finally中release了，就失去了传递性（当然，如果在这个ChannelHandler内这个ByteBuf已完成了它的使命，那么便无须再传递）

基本规则是，谁是最后使用者，谁负责release，详细分析如下：

  • 起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）

  • 入站 ByteBuf 处理原则

    • 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release

    • 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release

    • 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release

    • 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release

    • 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）

  • 出站 ByteBuf 处理原则

    • 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release

  • 异常处理原则

    • 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true

9）slice

【零拷贝】的体现之一，对原始ByteBuf进行切片成多个ByteBuf，切片后的ByteBuf并没有发生内存复制，还是使用原始ByteBuf的内存，切片后的ByteBuf维护独立的read和write指针。

示例1：

package cn.itcast.netty.c2;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;import static cn.itcast.netty.c2.TestByteBuf.log;public class TestSlice {public static void main(String[] args) {ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);buf.writeBytes(new byte[]{'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'});System.out.println("-----------------------有参slice--------------------------------");log(buf);// 在切换过程中没有发生数据复制ByteBuf f1 = buf.slice(0, 5);log(f1);ByteBuf f2 = buf.slice(5, 5);log(f2);System.out.println("----------------------更改分片内容---------------------------------");// 更改slice的内容，则原始ByteBuf也会受影响，因为底层都是同一款块内存f1.setByte(0,  'b');log(f1);log(buf);System.out.println("-----------------------无参slice--------------------------------");ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);origin.writeBytes(new byte[]{1, 2, 3, 4});log(origin);origin.readByte();log(origin);// 无参slice是从原始ByteBuf的read index到write index之间的内容进行切片，切片后的max capacity被固定为这个区间的大小，不能追加writeByteBuf slice = origin.slice();log(slice);
//        slice.writeBytes(5);  // 会报错，IndexOutOfBoundsException异常System.out.println("-----------------------原始ByteBuf再次读操作--------------------------------");origin.readByte();log(origin);log(slice);System.out.println("-----------------------释放原始ByteBuf的内存--------------------------------");slice.retain();origin.release();log(origin);  //  如果没有slice.retain() -> io.netty.util.IllegalReferenceCountExceptionlog(slice);}
}

输出

-----------------------有参slice--------------------------------
read index:0 write index:10 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 66 67 68 69 6a                   |abcdefghij      |
+--------+-------------------------------------------------+----------------+
read index:0 write index:5 capacity:5
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65                                  |abcde           |
+--------+-------------------------------------------------+----------------+
read index:0 write index:5 capacity:5
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 67 68 69 6a                                  |fghij           |
+--------+-------------------------------------------------+----------------+
----------------------更改分片内容---------------------------------
read index:0 write index:5 capacity:5
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 62 63 64 65                                  |bbcde           |
+--------+-------------------------------------------------+----------------+
read index:0 write index:10 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 62 63 64 65 66 67 68 69 6a                   |bbcdefghij      |
+--------+-------------------------------------------------+----------------+
-----------------------无参slice--------------------------------
read index:0 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04                                     |....            |
+--------+-------------------------------------------------+----------------+
read index:1 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+
read index:0 write index:3 capacity:3
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+
-----------------------原始ByteBuf再次读操作--------------------------------
read index:2 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04                                           |..              |
+--------+-------------------------------------------------+----------------+
read index:0 write index:3 capacity:3
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+
-----------------------释放原始ByteBuf的内存--------------------------------
read index:2 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04                                           |..              |
+--------+-------------------------------------------------+----------------+
read index:0 write index:3 capacity:3
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04                                        |...             |
+--------+-------------------------------------------------+----------------+

10）duplicate

【零拷贝】的体现之一，就好比截取了原始ByteBuf所有内容，并且没有max capacity的限制，也是与原始ByteBuf使用同一块底层内存，只是读写指针是独立的。

示例1：

package cn.itcast.netty.c2;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;import static cn.itcast.netty.c2.TestByteBuf.log;public class TestDuplicate {public static void main(String[] args) {ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);buf.writeBytes(new byte[]{'a', 'b', 'c', 'd'});log(buf);ByteBuf duplicate = buf.duplicate();log(duplicate);buf.writeBytes(new byte[]{'x', 'y'});log(buf);duplicate.writeBytes(new byte[]{'e', 'f', 'g', 'h', 'i', 'j', 'k'});log(buf);log(duplicate);}
}

输出

read index:0 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64                                     |abcd            |
+--------+-------------------------------------------------+----------------+
read index:0 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64                                     |abcd            |
+--------+-------------------------------------------------+----------------+
read index:0 write index:6 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 78 79                               |abcdxy          |
+--------+-------------------------------------------------+----------------+
read index:0 write index:6 capacity:64
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 66                               |abcdef          |
+--------+-------------------------------------------------+----------------+
read index:0 write index:11 capacity:64
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 66 67 68 69 6a 6b                |abcdefghijk     |
+--------+-------------------------------------------------+----------------+

11）copy

会将底层内存数据进行深拷贝，因此无论读写，都与原始ByteBuf无关。

12）CompositeByte

【零拷贝】的体现之一，可以将多个ByteBuf合并为一个逻辑上的ByteBuf，避免拷贝。

示例：

package cn.itcast.netty.c2;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;import static cn.itcast.netty.c2.TestByteBuf.log;public class TestCompositeByteBuf {public static void main(String[] args) {ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer();buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer();buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});/*ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();buffer.writeBytes(buf1).writeBytes(buf2);log(buffer);*/CompositeByteBuf buffer = ByteBufAllocator.DEFAULT.compositeBuffer();buffer.addComponents(true, buf1, buf2);log(buffer);}
}

输出

read index:0 write index:10 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
+--------+-------------------------------------------------+----------------+

CompositeByteBuf是一个组合的ByteBuf，它内部维护了一个Component数组，每个Component管理一个ByteBuf，记录了这个ByteBuf相对于整体偏移量等信息，代表着整体中某一段的数据。

  • 优点，对外是一个虚拟视图，组合这些ByteBuf不会产生内存复制；
  • 缺点，复杂了很多，多次操作会带来性能的损耗。

13）Unpooled

Unpooled是一个工具类，类如其名，提供了非池化的ByteBuf创建、组合、复制等操作。

这里仅介绍其跟【零拷贝】相关的wrappedBuffer方法，可以用来包装ByteBuf

package cn.itcast.netty.c2;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;import static cn.itcast.netty.c2.TestByteBuf.log;public class TestUnpooled {public static void main(String[] args) {ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});// 当包装ByteBuf个数超过1个时，底层使用了CompositeByteBufByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);log(buf3);// 也可以用来包装普通字节数组，底层也不会有拷贝操作ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});System.out.println(buf4.getClass());log(buf4);}
}

输出

read index:0 write index:10 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
+--------+-------------------------------------------------+----------------+
class io.netty.buffer.CompositeByteBuf
read index:0 write index:6 capacity:6
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06                               |......          |
+--------+-------------------------------------------------+----------------+

ByteBuf优势

  • 池化 - 可以重用池中ByteBuf实例，更节约内存，减少内存溢出的可能；
  • 读写指针分离，不需要像ByteBuffer一样切换读写模式；
  • 可以自动扩容
  • 支持链式调用，使用更流畅
  • 很多地方体现零拷贝，例如slice、duplicate、CompositeByteBuf

4. 双向通信

需求：实现一个echo server

服务器端：

package cn.itcast.netty.c2;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;import java.nio.charset.Charset;public class EchoServer {public static void main(String[] args) throws InterruptedException {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {try {ByteBuf buffer = (ByteBuf) msg;System.out.println("Server received: " + buffer.toString(Charset.defaultCharset()));// 创建响应并使用相同的消息内容ByteBuf reponse = ctx.alloc().buffer();reponse.writeBytes(buffer);ctx.writeAndFlush(reponse);// 注意：这里不需要手动释放buffer，因为Netty会在pipeline处理完成后自动释放} catch (Exception e) {e.printStackTrace();ctx.close();}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}});}}).bind(8080).sync().channel().closeFuture().sync();}
}

客户端：

package cn.itcast.netty.c2;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.ReferenceCountUtil;import java.nio.charset.Charset;
import java.util.Scanner;public class EchoClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();try {Channel channel = new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder(Charset.defaultCharset()));ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {try {ByteBuf buffer = (ByteBuf) msg;System.out.println("Client received: " + buffer.toString(Charset.defaultCharset()));// 注意：这里不需要手动释放buffer，因为Netty会在pipeline处理完成后自动释放} finally {ReferenceCountUtil.release(msg);  // 更安全的做法}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}});}}).connect("localhost", 8080).sync().channel();channel.closeFuture().addListener(future -> {group.shutdownGracefully();  // 优雅关闭});Scanner sc = new Scanner(System.in);while (true) {String line = sc.nextLine();if("q".equals(line)) {channel.close();break;}channel.writeAndFlush(line);}} finally {group.shutdownGracefully();  // 优雅关闭}}
}

客户端输出

hello server
Client received: hello server
hello, world!
Client received: hello, world!
q

服务器端输出

Server received: hello server
Server received: hello, world!

关键点说明

1. 资源释放问题：

  • 在简单的echo服务中，通常不需要手动释放ByteBuf，因为Netty会自动管理接收到的消息的内存。
  • 但最佳实践是：
    • 如果你消费了消息（如转发了它），则不需要释放；
    • 如果你没有消费消息，应该调用ReferenceCountUtil.release(msg)
    • 或者在handler前添加SimpleChannelInboundHandler，它会自动释放

2. 异常处理：

  • 添加了exceptionCaught方法来处理异常情况
  • 确保在异常时关闭连接

3. 双向通信：

  • 服务器在收到消息后会原样返回
  • 客户端可以发送消息并接收服务器的响应

4. 优雅关闭：

  • 使用shutdownGracefully()确保资源正确释放
  • 添加了关闭监听器

5. 改进建议：

  • 可以使用SimpleChannelInboundHandler替代ChannelInboundHandlerAdapter简化代码
  • 可以添加空闲状态检测
  • 可以添加日志记录。

相关文章：

Netty从入门到进阶（二）

二、Netty入门 1. 概述 1.1 Netty是什么 Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients. Netty是一个异步的、基于事件驱动的网络应用框架&#xff0c;用于…...

编程日记 2025/6/12 8:42:58

深入浅出深度学习基础：从感知机到全连接神经网络的核心原理与应用

文章目录 前言一、感知机 (Perceptron)1.1 基础介绍1.1.1 感知机是什么&#xff1f;1.1.2 感知机的工作原理 1.2 感知机的简单应用&#xff1a;基本逻辑门1.2.1 逻辑与 (Logic AND)1.2.2 逻辑或 (Logic OR)1.2.3 逻辑与非 (Logic NAND) 1.3 感知机的实现1.3.1 简单实现 (基于阈…...

编程日记 2025/6/12 4:39:25

SQL慢可能是触发了ring buffer

简介 最近在进行 postgresql 性能排查的时候，发现 PG 在某一个时间并行执行的 SQL 变得特别慢。最后通过监控监观察到并行发起得时间 buffers_alloc 就急速上升，且低水位伴随在整个慢 SQL，一直是 buferIO 的等待事件，此时也没有其他会话的争抢。SQL 虽然不是高效 SQL ，但…...

编程日记 2025/6/12 5:36:18

C#中的CLR属性、依赖属性与附加属性

CLR属性的主要特征 封装性&#xff1a; 隐藏字段的实现细节 提供对字段的受控访问 访问控制&#xff1a; 可单独设置get/set访问器的可见性 可创建只读或只写属性 计算属性&#xff1a; 可以在getter中执行计算逻辑 不需要直接对应一个字段 验证逻辑&#xff1a; 可以…...

编程日记 2025/6/12 2:04:37

使用LangGraph和LangSmith构建多智能体人工智能系统

现在&#xff0c;通过组合几个较小的子智能体来创建一个强大的人工智能智能体正成为一种趋势。但这也带来了一些挑战&#xff0c;比如减少幻觉、管理对话流程、在测试期间留意智能体的工作方式、允许人工介入以及评估其性能。你需要进行大量的反复试验。 在这篇博客〔原作者&a…...

编程日记 2025/6/12 8:04:49

MySQL 知识小结（一）

一、my.cnf配置详解 我们知道安装MySQL有两种方式来安装咱们的MySQL数据库&#xff0c;分别是二进制安装编译数据库或者使用三方yum来进行安装,第三方yum的安装相对于二进制压缩包的安装更快捷&#xff0c;但是文件存放起来数据比较冗余&#xff0c;用二进制能够更好管理咱们M…...

编程日记 2025/6/12 3:26:05

无人机侦测与反制技术的进展与应用

国家电网无人机侦测与反制技术的进展与应用 引言 随着无人机&#xff08;无人驾驶飞行器&#xff0c;UAV&#xff09;技术的快速发展&#xff0c;其在商业、娱乐和军事领域的广泛应用带来了新的安全挑战。特别是对于关键基础设施如电力系统&#xff0c;无人机的“黑飞”&…...

编程日记 2025/6/11 1:19:57

【SSH疑难排查】轻松解决新版OpenSSH连接旧服务器的“no matching...“系列算法协商失败问题

【SSH疑难排查】轻松解决新版OpenSSH连接旧服务器的"no matching..."系列算法协商失败问题 摘要&#xff1a; 近期&#xff0c;在使用较新版本的OpenSSH客户端连接老旧SSH服务器时&#xff0c;会遇到 "no matching key exchange method found"​, "n…...

编程日记 2025/6/11 1:18:52

【无标题】路径问题的革命性重构：基于二维拓扑收缩色动力学模型的零点隧穿理论

路径问题的革命性重构&#xff1a;基于二维拓扑收缩色动力学模型的零点隧穿理论 一、传统路径模型的根本缺陷 在经典正方形路径问题中&#xff08;图1&#xff09;&#xff1a; mermaid graph LR A((A)) --- B((B)) B --- C((C)) C --- D((D)) D --- A A -.- C[无直接路径] B -…...

编程日记 2025/6/12 5:06:10

Go 并发编程基础：通道（Channel）的使用

在 Go 中&#xff0c;Channel 是 Goroutine 之间通信的核心机制。它提供了一个线程安全的通信方式&#xff0c;用于在多个 Goroutine 之间传递数据&#xff0c;从而实现高效的并发编程。 本章将介绍 Channel 的基本概念、用法、缓冲、关闭机制以及 select 的使用。 一、Channel…...

编程日记 2025/6/11 1:16:42

免费PDF转图片工具

免费PDF转图片工具 一款简单易用的PDF转图片工具&#xff0c;可以将PDF文件快速转换为高质量PNG图片。无需安装复杂的软件&#xff0c;也不需要在线上传文件&#xff0c;保护您的隐私。 工具截图 主要特点 &#x1f680; 快速转换&#xff1a;本地转换&#xff0c;无需等待上…...

编程日记 2025/6/12 8:02:23

Kafka入门-生产者

生产者 生产者发送流程&#xff1a; 延迟时间为0ms时&#xff0c;也就意味着每当有数据就会直接发送 异步发送API 异步发送和同步发送的不同在于&#xff1a;异步发送不需要等待结果&#xff0c;同步发送必须等待结果才能进行下一步发送。 普通异步发送 首先导入所需的k…...

编程日记 2025/6/12 8:15:06

深度学习水论文：mamba＋图像增强

&#x1f9c0;当前视觉领域对高效长序列建模需求激增&#xff0c;对Mamba图像增强这方向的研究自然也逐渐火热。原因在于其高效长程建模&#xff0c;以及动态计算优势&#xff0c;在图像质量提升和细节恢复方面有难以替代的作用。 &#x1f9c0;因此短时间内&#xff0c;就有不…...

编程日记 2025/6/12 8:48:11

AirSim/Cosys-AirSim 游戏开发（四）外部固定位置监控相机

这个博客介绍了如何通过 settings.json 文件添加一个无人机外的 固定位置监控相机&#xff0c;因为在使用过程中发现 Airsim 对外部监控相机的描述模糊&#xff0c;而 Cosys-Airsim 在官方文档中没有提供外部监控相机设置&#xff0c;最后在源码示例中找到了&#xff0c;所以感…...

编程日记 2025/6/12 7:50:20

现有的 Redis 分布式锁库（如 Redisson）提供了哪些便利？

现有的 Redis 分布式锁库&#xff08;如 Redisson&#xff09;相比于开发者自己基于 Redis 命令&#xff08;如 SETNX, EXPIRE, DEL&#xff09;手动实现分布式锁&#xff0c;提供了巨大的便利性和健壮性。主要体现在以下几个方面&#xff1a; 原子性保证 (Atomicity)&#xff…...

编程日记 2025/6/12 8:37:34

【Redis】笔记｜第8节｜大厂高并发缓存架构实战与优化

缓存架构 代码结构 代码详情 功能点&#xff1a; 多级缓存&#xff0c;先查本地缓存&#xff0c;再查Redis&#xff0c;最后才查数据库热点数据重建逻辑使用分布式锁&#xff0c;二次查询更新缓存采用读写锁提升性能采用Redis的发布订阅机制通知所有实例更新本地缓存适用读多…...

编程日记 2025/6/12 8:48:12

C/C++ 中附加包含目录、附加库目录与附加依赖项详解

在 C/C 编程的编译和链接过程中&#xff0c;附加包含目录、附加库目录和附加依赖项是三个至关重要的设置&#xff0c;它们相互配合&#xff0c;确保程序能够正确引用外部资源并顺利构建。虽然在学习过程中&#xff0c;这些概念容易让人混淆&#xff0c;但深入理解它们的作用和联…...

编程日记 2025/6/12 1:28:45

【从零学习JVM|第三篇】类的生命周期(高频面试题)

前言&#xff1a; 在Java编程中&#xff0c;类的生命周期是指类从被加载到内存中开始&#xff0c;到被卸载出内存为止的整个过程。了解类的生命周期对于理解Java程序的运行机制以及性能优化非常重要。本文会深入探寻类的生命周期&#xff0c;让读者对此有深刻印象。 目录 ​…...

编程日记 2025/6/11 1:06:04

Mysql8 忘记密码重置，以及问题解决

1.使用免密登录 找到配置MySQL文件&#xff0c;我的文件路径是/etc/mysql/my.cnf&#xff0c;有的人的是/etc/mysql/mysql.cnf 在里最后加入 skip-grant-tables重启MySQL服务 service mysql restartShutting down MySQL… SUCCESS! Starting MySQL… SUCCESS! 重启成功 2.登…...

编程日记 2025/6/12 4:21:57

C++.OpenGL （20/64）混合（Blending）

混合（Blending） 透明效果核心原理 #mermaid-svg-SWG0UzVfJms7Sm3e {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-SWG0UzVfJms7Sm3e .error-icon{fill:#552222;}#mermaid-svg-SWG0UzVfJms7Sm3e .error-text{fill…...

编程日记 2025/6/12 6:57:13

Linux 中如何提取压缩文件 ？

Linux 是一种流行的开源操作系统&#xff0c;它提供了许多工具来管理、压缩和解压缩文件。压缩文件有助于节省存储空间&#xff0c;使数据传输更快。本指南将向您展示如何在 Linux 中提取不同类型的压缩文件。 1. Unpacking ZIP Files ZIP 文件是非常常见的&#xff0c;要在 …...

编程日记 2025/6/12 6:29:09

纯 Java 项目（非 SpringBoot）集成 Mybatis-Plus 和 Mybatis-Plus-Join

纯 Java 项目&#xff08;非 SpringBoot&#xff09;集成 Mybatis-Plus 和 Mybatis-Plus-Join 1、依赖1.1、依赖版本1.2、pom.xml 2、代码2.1、SqlSession 构造器2.2、MybatisPlus代码生成器2.3、获取 config.yml 配置2.3.1、config.yml2.3.2、项目配置类 2.4、ftl 模板2.4.1、…...

编程日记 2025/6/12 8:02:24

Java求职者面试指南：计算机基础与源码原理深度解析

Java求职者面试指南&#xff1a;计算机基础与源码原理深度解析 第一轮提问&#xff1a;基础概念问题 1. 请解释什么是进程和线程的区别&#xff1f; 面试官&#xff1a;进程是程序的一次执行过程&#xff0c;是系统进行资源分配和调度的基本单位&#xff1b;而线程是进程中的…...

编程日记 2025/6/12 6:26:22

Java数值运算常见陷阱与规避方法

整数除法中的舍入问题 问题现象 当开发者预期进行浮点除法却误用整数除法时，会出现小数部分被截断的情况。典型错误模式如下： void process(int value) {double half = value / 2; // 整数除法导致截断// 使用half变量 }此时...

编程日记 2025/6/12 6:26:12

GitFlow 工作模式（详解）

今天再学项目的过程中遇到使用gitflow模式管理代码&#xff0c;因此进行学习并且发布关于gitflow的一些思考 Git与GitFlow模式 我们在写代码的时候通常会进行网上保存&#xff0c;无论是github还是gittee&#xff0c;都是一种基于git去保存代码的形式&#xff0c;这样保存代码…...

编程日记 2025/6/12 8:17:34

AI+无人机如何守护濒危物种？YOLOv8实现95%精准识别

【导读】 野生动物监测在理解和保护生态系统中发挥着至关重要的作用。然而&#xff0c;传统的野生动物观察方法往往耗时耗力、成本高昂且范围有限。无人机的出现为野生动物监测提供了有前景的替代方案&#xff0c;能够实现大范围覆盖并远程采集数据。尽管具备这些优势&#xf…...

编程日记 2025/6/12 12:44:46

MFC 抛体运动模拟：常见问题解决与界面美化

在 MFC 中开发抛体运动模拟程序时，我们常遇到 轨迹残留、无效刷新、视觉单调、物理逻辑瑕疵 等问题。本文将针对这些痛点，详细解析原因并提供解决方案，同时兼顾界面美化，让模拟效果更专业、更高效。 问题一：历史轨迹与小球残影残留 现象 小球运动后，历史位置的 “残影”…...

编程日记 2025/6/12 3:01:17

云原生安全实战：API网关Kong的鉴权与限流详解

&#x1f525;「炎码工坊」技术弹药已装填&#xff01; 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 一、基础概念 1. API网关&#xff08;API Gateway&#xff09; API网关是微服务架构中的核心组件&#xff0c;负责统一管理所有API的流量入口。它像一座…...

编程日记 2025/6/12 13:07:36

【C++特殊工具与技术】优化内存分配(一)：C++中的内存分配

目录 一、C 内存的基本概念​ 1.1 内存的物理与逻辑结构​ 1.2 C 程序的内存区域划分​ 二、栈内存分配​ 2.1 栈内存的特点​ 2.2 栈内存分配示例​ 三、堆内存分配​ 3.1 new和delete操作符​ 4.2 内存泄漏与悬空指针问题​ 4.3 new和delete的重载​ 四、智能指针…...

编程日记 2025/6/12 13:12:23

vulnyx Blogger writeup

信息收集 arp-scan nmap 获取userFlag 上web看看 一个默认的页面&#xff0c;gobuster扫一下目录 可以看到扫出的目录中得到了一个有价值的目录/wordpress&#xff0c;说明目标所使用的cms是wordpress&#xff0c;访问http://192.168.43.213/wordpress/然后查看源码能看到 这…...

编程日记 2025/6/12 13:06:34