PatrickGopher

@Override
public void getUserInfo(UserInfoRequest request, StreamObserver<UserInfo> response) {
  final CompletableFuture<String> Result =
      provider.getUserInfo(request.getUserId()); // some other grpc request invoke
  final CompletableFuture<UserInfo> responseFuture =
      trackResult.thenApply(result -> UserInfo.newBuilder().setUserInfo(result).build());
  responseFuture.whenComplete(
      (result, throwable) -> {
        if (result != null) {
          response.onNext(result);

PatrickGopher

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor)
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor)

PatrickGopher

final var a = 10;
CompletableFuture<Void> future = CompletableFuture
    .supplyAsync(() ->  a * (a + 1))
    .exceptionally(ex -> {
      System.out.println("err: " + ex.toString());
      return 0;
    })
    .thenApply(i -> Integer.toString(i))
    .thenApply(str -> "\"" + str + "\"")
    .thenAccept(System.out::println);

PatrickGopher

ExecutorService executor = Executors.newFixedThreadPool(5);
List<Future<String>> list = new ArrayList<>();
for(int i=0; i< 100; i++){
  Future<String> future = executor.submit(() -> "222");
  list.add(future);
}
for(Future<String> fut : list){
  try {
    System.out.println(new Date()+ "::"+fut.get());
  } catch (InterruptedException | ExecutionException e) {

PatrickGopher

ExecutorService executor = Executors.newFixedThreadPool(5);
List<Future<String>> list = new ArrayList<>();
for(int i=0; i< 100; i++){
  Future<String> future = executor.submit(() -> "222");
  list.add(future);
}
for(Future<String> fut : list){
  try {
    System.out.println(new Date()+ "::"+fut.get());
  } catch (InterruptedException | ExecutionException e) {

PatrickGopher

public static void main(String[] args) {
        CompletableFuture<Void> subTask = null;
        try (SubmissionPublisher<Long> pub = new SubmissionPublisher<>()) {
            // Create a Subscriber
            SimpleSubscriber sub = new SimpleSubscriber("S1", 10);
            // Create a processor
            FilterProcessor<Long> filter = new FilterProcessor<>(n -> n % 2 == 1);
            // Subscribe the filter to the publisher and a subscriber to the filter
            pub.subscribe(filter);  // publisher add processer          
            filter.subscribe(sub);  // processer add subscriber

PatrickGopher

public class NormalSubscriber implements Flow.Subscriber<Long> {    
    private Flow.Subscription subscription;
    private String name = "NormalSubscriber";
    public NormalSubscriber(String name) {
        this.name = name;
    }
    public String getName() {
        return name;
    }
    @Override

PatrickGopher

public class FilterProcessor<T> extends SubmissionPublisher<T> implements Processor<T,T>{
    private Predicate<? super T> filter;
    public FilterProcessor(Predicate<? super T> filter) {
        this.filter = filter;
    }
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        subscription.request(5);// set up your queue length
    }
    @Override

PatrickGopher

func delay(ctx context.Context, t int) {
	go func() {
		for {
			time.Sleep(2 * time.Second)
			fmt.Println("running ... ")
		}
	}()

	select {
	case <-ctx.Done():

PatrickGopher

func task(ctx context.Context, cnt int) {
	for {
		select {
		case v := <-ctx.Done():
			fmt.Printf("counter：%v，receive: %v, finish\n", cnt, v)
			return
		default:
			fmt.Printf("counter：%v，finish\n", cnt)
			time.Sleep(1 * time.Second)
		}