public interface Callback<T> {
    void onSuccess(T t);
    void onError(String error);
}

public class CallbackUserService {
    public void getFavorites(Long userId, Callback<List<Long>> callback) {
        new Thread(() -> {
            try {
                // 模擬數(shù)據(jù)庫訪問時間
                Thread.sleep(1000);
                List<Long> favorites = Arrays.asList(1L,2L,3L);
                callback.onSuccess(favorites);
            } catch (Exception e) {
                callback.onError("讀取出現(xiàn)錯誤");
            }
        }).start();
    }
}

public class CallbackFavoriteService {
    private Map<Long, String> names = new HashMap<Long, String>() {{
        put(1L, "football");
        put(2L, "movie");
        put(3L, "film");
    }};
    public void getDetail(Long id, Callback<String> callback) {
        new Thread(() -> {
            try {
                // 模擬數(shù)據(jù)庫訪問時間
                Thread.sleep(1000);
                callback.onSuccess(names.get(id));
            } catch (Exception e) {
                callback.onError("讀取出現(xiàn)錯誤");
            }
        }).start();
    }
}

userService.getFavorites(23L, new Callback<List<Long>>() {
    @Override
    public void onSuccess(List<Long> favIds) {
        favIds.stream().limit(2).forEach(favId->{
            favoriteService.getDetail(favId, new Callback<String>() {
                @Override
                public void onSuccess(String s) {
                    System.out.println(s);
                }
                @Override
                public void onError(String error) {
                    System.out.println(error);
                }
            });
        });
    }
    @Override
    public void onError(String error) {
        System.out.println(error);
    }
});
while (true) {
    System.out.println("做點(diǎn)其它事。。。");
    Thread.sleep(1000);
}

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.3.8.RELEASE</version>
</dependency>

public class ReactorUserService {
    // 返回的是多個，即Flux
    public Flux<Long> getFavorites(Long userId) {
        // 創(chuàng)建Flux并返回
        return Flux.create(sink -> {
            new Thread(() -> {
                // 模擬數(shù)據(jù)庫訪問時間
                try {
                    Thread.sleep(1000);
                    // 發(fā)布數(shù)據(jù) 1,2,3
                    sink.next(1L);
                    sink.next(2L);
                    sink.next(3L);
                    // 標(biāo)識發(fā)布結(jié)束
                    sink.complete();
                } catch (Exception e) {
                    e.printStackTrace();
                    sink.error(new Exception("讀取出現(xiàn)錯誤"));
                }
            }).start();
        });
    }
}

public class ReactorFavoriteService {
    private Map<Long, String> names = new HashMap<Long, String>() {{
        put(1L, "football");
        put(2L, "movie");
        put(3L, "film");
    }};
    // 返回單個數(shù)據(jù)，所以是Mono
    public Mono<String> getDetail(Long id) {
        return Mono.create(sink -> {
            new Thread(() -> {
                // 模擬數(shù)據(jù)庫訪問時間
                try {
                    Thread.sleep(1000);
                    // 發(fā)布并完成
                    sink.success(names.get(id));
                } catch (Exception e) {
                    e.printStackTrace();
                    sink.error(new Exception("讀取出現(xiàn)錯誤"));
                }
            }).start();
        });
    }
}

userService.getFavorites(23L)
        .flatMap(id->favoriteService.getDetail(id)) // 取詳情
        .take(2) // 取前兩個
        .subscribe(System.out::println, error-> System.out.println("process error:"+error)); // 訂閱，處理方式sout
while (true) {
    System.out.println("做點(diǎn)其它事。。。");
    Thread.sleep(1000);
}

public class ReactorRedisSuggestionService {
    private RedisURI redisUri = RedisURI.builder().withHost("127.0.0.1").withPort(6379).build();
    public Mono<String> getSuggestions() {
        return Mono.create(sink -> {
            RedisClient redisClient = RedisClient.create(redisUri); // 客戶端
            StatefulRedisConnection<String, String> connection = redisClient.connect(); // 連接
            RedisAsyncCommands<String, String> asyncCommands = connection.async(); // 異步指令
            asyncCommands.get("favorites").thenAccept(favorites->{// 異步獲取，key為favorites
                sink.success(favorites); // 返回數(shù)據(jù)后推送給mono
                connection.close();  // 關(guān)閉連接
                redisClient.shutdown();
            });
        });
    }
}

suggestionService.getSuggestions()
                .subscribe(System.out::println, error-> System.out.println("process error:"+error));

public Mono<String> getSuggestions() {
        RedisReactiveCommands<String, String> reactiveCommands = connection.reactive(); // 響應(yīng)式指令
        return reactiveCommands.get("favorites"); // key為favorites，返回的就是Mono
}

@Test
public void sync() {
    Executor pool = Executors.newFixedThreadPool(10); // 請求處理線程 10個
    RedisClient redisClient = RedisClient.create(redisUri);
    StatefulRedisConnection<String, String> connection = redisClient.connect();
    long startTime = System.currentTimeMillis();
    for (int i=0;i<30;i++) { // 模擬30個請求
        pool.execute(()->{
            connection.sync().get("favorites"); // 同步獲取
        });
    }
    pool.execute(()->{
        // 執(zhí)行到這里代表線程池的線程都釋放出來了，可以做其它事情了，記錄一下時間
        long endTIme = System.currentTimeMillis();
        System.out.println("free:" + (endTIme-startTime));
    });
    for(;;);
}

@Test
public void async() {
    Executor pool = Executors.newFixedThreadPool(5); // 請求處理線程 5個
    ClientResources res = DefaultClientResources.builder().ioThreadPoolSize(5).build(); // 回調(diào)處理線程5個
    RedisClient redisClient = RedisClient.create(res, redisUri);
    StatefulRedisConnection<String, String> connection = redisClient.connect();
    long startTime = System.currentTimeMillis();
    for (int i=0;i<30;i++) { // 模擬30個請求
        pool.execute(()-> {
            RedisReactiveCommands<String, String> reactiveCommands = connection.reactive();
            reactiveCommands.get("favorites").subscribe(); // 異步獲取
        });
    }
    // 執(zhí)行到這里代表線程池的線程都釋放出來了，可以做其它事情了，記錄一下時間
    pool.execute(()->{
        long endTIme = System.currentTimeMillis();
        System.out.println("free:" + (endTIme-startTime));
    });
    for(;;);
}