public class TimeServer {
public void bind(int port) {
try {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel arg0) throws Exception {
System.out.println("初始化");
arg0.pipeline().addLast(new TimeHandler());
}
});
ChannelFuture future = b.bind(port).sync();
System.out.println("执行这里");
future.channel().closeFuture().sync();
System.out.println("执行这里");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
new TimeServer().bind(10000);
}
}
public class TimeHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
super.channelRead(ctx, msg);
ByteBuf buf=(ByteBuf) msg;
//// byte[] butfs = buf.array();//报错
System.out.println(buf.readableBytes());
byte[] butfs = new byte[buf.readableBytes()];
buf.readBytes(butfs);
System.out.println(new String(butfs,"UTF-8"));
System.out.println(msg);
}
}
客户端使用的是BIO的模型：
public static void main(String[] args) throws Exception {
final int port = 10000;
// NioServer server = new NioServer(port);
// server.init();
/// ========================================================
// 接下来模拟3个Client并发访问服务器
int poolsize = 1;
ExecutorService pool = Executors.newFixedThreadPool(poolsize);
Collection<Callable> tasks = new ArrayList<Callable>(10);
final String clientname = "clientThread";
for (int i = 0; i < poolsize; i++) {
final int n = i;
// 若每一个Client都保持使用BIO方式发送数据到Server，并读取数据。
tasks.add(new Callable() {
@Override
public Object call() throws Exception {
Socket socket = new Socket("127.0.0.1", port);
final InputStream input = socket.getInputStream();
final OutputStream out = socket.getOutputStream();
final String clientname_n = clientname + "_" + n;
// BIO读取数据线程
new Thread(clientname_n + "_read") {
@Override
public void run() {
byte[] bs = new byte[1024];
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
int len = 0;
try {
while ((len = input.read(bs)) != -1) {
System.out.println("Clinet thread " + Thread.currentThread().getName()
+ " read: " + new String(bs, 0, len));
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}.start();
// BIO写数据线程
new Thread(clientname_n + "_write") {
@Override
public void run() {
int a = 0;
while (true) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
String str = Thread.currentThread().getName() + " hello, " + a;
try {
out.write(str.getBytes());
out.flush();
a++;
} catch (IOException e) {
e.printStackTrace();
}
}
}
}.start();
return null;
}
});
}
pool.invokeAll((Collection<? extends Callable<Object>>) tasks);
// server.go();
}结果运行的时候出现了以下错误：
月 13, 2017 5:52:56 下午 io.netty.channel.DefaultChannelPipeline onUnhandledInboundException
警告: An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
io.netty.util.IllegalReferenceCountException: refCnt: 0
at io.netty.buffer.AbstractByteBuf.ensureAccessible(AbstractByteBuf.java:1408)
at io.netty.buffer.AbstractByteBuf.checkReadableBytes0(AbstractByteBuf.java:1394)
at io.netty.buffer.AbstractByteBuf.checkReadableBytes(AbstractByteBuf.java:1383)
at io.netty.buffer.AbstractByteBuf.readBytes(AbstractByteBuf.java:850)
at io.netty.buffer.AbstractByteBuf.readBytes(AbstractByteBuf.java:858)
at test.netty.TimeHandler.channelRead(TimeHandler.java:17)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:134)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:624)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:559)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:476)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:438)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
at java.lang.Thread.run(Thread.java:745)
请问这是为什么呢？
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
netty4에서는 객체의 수명 주기가 참조 카운터에 의해 제어됩니다. ByteBuf의 경우도 마찬가지입니다. 릴리스 메서드가 한 번 호출되면 참조 카운터가 감소합니다. by 1. 카운터가 0인 객체에 접근하려고 하면, verifyAccessible 구현과 마찬가지로 IllegalReferenceCountException이 발생합니다. 자세한 설명은 공식 문서
를 참조하세요.AbstractByteBuf.java
으아악TZ의 TimeHandler 클래스에 있는 super.channelRead(ctx, msg); 코드 줄에 주의하세요. 호출 경로를 추적하세요.
으아악호출된 최종 코드는: ReferenceCountUtil.release(msg)
으아악즉, super.channelRead(ctx, msg);가 호출될 때마다 ByteBuf는 release() 메서드를 호출하고 카운터는 1씩 감소한 다음 buf.readBytes(butfs)에서 이 코드 줄은 다음과 같이 됩니다. verifyAccessible() 을 확인하면 카운터는 0이고 netty는 ByteBuf 객체가 해제되었다고 생각하고 예외를 발생시킵니다.
해결책:
TimeHandler super.channelRead(ctx, msg);
ByteBuf 객체를 처리하는 사람이 이를 해제합니다.
ExceptionCaught() 이벤트가 발생하여 파이프라인 끝에 도달했습니다. 이는 일반적으로 파이프라인의 마지막 처리기가 예외를 처리하지 않았음을 의미합니다.
缺少ExceptionCaught(),재재서버端最后一个Handler中增加ExceptionCaught()