[Java知识库]netty之future和promise详解

Future和Promise

jdk Future：只能同步等待任务结束才能得到结果
netty Future： 可以同步或者异步等待任务结束之后得到结果，继承自jdk Future。
netty promise：脱离了任务而独立存在，作为两个线程间传递结果的容器，继承自netty Future。

JDK Future

一个例子：

package com.throne.netty.test1.futurepromise;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * @version v1.0
 * @Date: 2022/4/5 20:19
 * @Author: Mr.Throne
 * @Description:
 */
@Slf4j
public class TestJdkPromise {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 创建线程池
        ExecutorService service = Executors.newFixedThreadPool(2);

        // 提交任务
        Future<Integer> future = service.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Thread.sleep(1000);
                log.info("{}","子线程");
                return 100;
            }
        });

        log.info("{}","等待结果...");
        // 主线程通过future获取结果, 阻塞等待结果返回
        Integer integer = future.get();
        log.info("result:{}",integer);

    }
}

执行结果，可以看到它是阻塞并等待结果返回的：

netty Future

package com.throne.netty.test1.futurepromise;

import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoop;
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.Callable;
import java.util.concurrent.ExecutionException;

/**
 * @version v1.0
 * @Date: 2022/4/5 20:25
 * @Author: Mr.Throne
 * @Description:
 */
@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(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.info("计算中...");
                Thread.sleep(1000);
                return 100;
            }
        });

        // 同步
//        log.info("等待结果...");
//        log.info("{}",future.get());

        //异步
        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                log.info("result:{}",future.getNow());
            }
        });

    }
}

这里主要给出异步调用的方法，采用listener实现：

netty Promise

个人感觉promise与netty future获取结果方式类似，不同之处在于可以手动设定success的结果和出错的fail结果。

package com.throne.netty.test1.futurepromise;

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;

/**
 * @version v1.0
 * @Date: 2022/4/5 20:31
 * @Author: Mr.Throne
 * @Description:
 */
@Slf4j
public class TestNettyPromise {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //获取一个eventLoop
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        EventLoop loop = eventLoopGroup.next();

        // 主动创建Promise,结果容器
        DefaultPromise<Integer> promise = new DefaultPromise<>(loop);

        new Thread(()->{
            // 由其他线程向promise填充结果
            log.info("开始计算...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            promise.setSuccess(100);
        }).start();

        //接收结果的线程
        log.info("waiting...");
        log.info("result:{}",promise.get());
        
    }
}