netty中Future和ChannelHandler
netty中的Future，继承自 jdk中的Future，， jdk中的Future，很垃圾，只能同步阻塞获取结果，，，
netty中的Future进行了升级，，可以addListener()异步获取结果，，可以isSuccess()判断任务成功还是失败，，
- jdk的Future
- get()
- isDone()
- cancel() : 取消当前任务
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(2);
Future<Integer> future = service.submit(() -> {
log.debug("running...");
Thread.sleep(2000);
return 2;
});
Integer i = future.get();
log.debug("i = " + i);
}
- netty中的Future
- isSuccess() : 判断任务是否成功
- sync() : 同步等待，，任务不成功会抛错
- getNow() : 获取结果，没有就返回null
- await() : 同步等待，，任务不成功不会报错，，后面通过isSuccess()判断是否成功
- addListener() : 任务结束回调
public static void main(String[] args) {
// netty中的线程池 eventLoop，， eventloop中就一个线程
NioEventLoopGroup group = new NioEventLoopGroup(2);
EventLoop eventLoop = group.next();
Future<String> future = eventLoop.submit(() -> {
Thread.sleep(2000);
return "hehe";
});
String now = future.getNow();
System.out.println("now = " + now);
boolean success = future.isSuccess();
System.out.println("success = " + success);
future.addListener(new GenericFutureListener<Future<? super String>>() {
@Override
public void operationComplete(Future<? super String> future) throws Exception {
Object now1 = future.getNow();
System.out.println("now1 = " + now1);
boolean success = future.isSuccess();
System.out.println("success = " + success);
}
});
}
- netty中的Promise
继承自netty的Future,
Promise可以设置成功和失败，，不用等到任务结束
public static void main(String[] args) throws ExecutionException, InterruptedException {
EventLoopGroup group = new NioEventLoopGroup(2);
EventLoop eventLoop = group.next();
// 主动创建promise ===> 结果的容器，
DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);
new Thread(()->{
System.out.println("开始计算");
try {
int i = 1/0;
Thread.sleep(1000);
promise.setSuccess(1000);
} catch (InterruptedException e) {
e.printStackTrace();
promise.setFailure(e);
}
//
}).start();
Integer i = promise.get();
System.out.println("i = " + i);
}
ChannelHandler
netty中handler分为两类：
- ChannelInboundHandler : 入站，， 读取数据，，，channel按照添加顺序依次执行
- ChannelOutboundHandler ：出站 ： 发送数据，，channel 逆序执行
channel.wirte() : 从末尾逆序执行
ctx.wirte() : 是从当前的handler，往前面找ChannelOutboundHandler执行
public static void main(String[] args) {
new ServerBootstrap()
.group(new NioEventLoopGroup(2))
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 添加处理器 ,,, netty会自动添加两个handler，，一个叫head，，一个叫tail，，，
// 底层是 双向链表
pipeline.addLast("handle01",new ChannelInboundHandlerAdapter(){
// 入站的handler，，一般关心的 read
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("msg:{}",msg);
ByteBuf byteBuf = (ByteBuf) msg;
String s = byteBuf.toString(Charset.defaultCharset());
// 调用下一个handle ctx.fireChannelRead(msg);，，并且将处理完成的结果，传递给下一个handler
super.channelRead(ctx, s);
}
});
pipeline.addLast("handle02",new ChannelInboundHandlerAdapter(){
// 入站的handler，，一般关心的 read
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("msg222:{}",msg);
User user = new User();
user.setName(((String) msg));
super.channelRead(ctx, user);
}
});
pipeline.addLast("handle03",new ChannelInboundHandlerAdapter(){
// 入站的handler，，一般关心的 read
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("msg333:{}",msg);
super.channelRead(ctx, msg);
ch.writeAndFlush(ctx.alloc().buffer().writeBytes("server".getBytes()));
}
});
// 出站是，，从后面往前走 ，，只有有写出的时候，才会触发出站方法，，，，
pipeline.addLast("handle04",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.debug("msg444:{}",msg);
super.write(ctx, msg, promise);
}
});
pipeline.addLast("handle05",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.debug("msg555:{}",msg);
super.write(ctx, msg, promise);
}
});
}
}).bind(new InetSocketAddress(8080));
}
EmbeddedChannel 模拟channel执行
public static void main(String[] args) {
ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("msg = " + msg);
super.channelRead(ctx, msg);
}
};
ChannelOutboundHandlerAdapter h2 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext channelHandlerContext, Object o, ChannelPromise channelPromise) throws Exception {
System.out.println(4444);
}
};
EmbeddedChannel channel = new EmbeddedChannel(h1,h2);
// channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hehe".getBytes()));
channel.writeOutbound(channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hehe".getBytes())));
}