【CompletableFuture任务编排】游戏服务器线程模型及其线程之间的交互(以排行榜线程和玩家线程的交互为例子)
需求：
1.我们希望玩家的业务在玩家线程执行，无需回调，因此是多线程处理。
2.匹配线程负责匹配逻辑，是单独一个线程。
3.排行榜线程负责玩家的上榜等。
4.从排行榜线程获取到排行榜列表后，需要给玩家发奖修改玩家数据，因此涉及到排行榜线程和玩家线程的交互。
5.房间线程也希望有多个，这样子各个房间之间业务无交互，进行并行执行。
ThreadManager.java // 负责所有线程的创建
package org.example.testLogicAndRank;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ThreadManager {/*** 逻辑专用线程*/public static ExecutorService[] logicThreadArr;/*** 房间专用线程*/public static ExecutorService[] roomThreadArr;/*** 排行榜专用线程*/public static ExecutorService rankExecutorService = Executors.newSingleThreadExecutor(r -> {Thread t = new Thread(r);t.setName("RankThread");return t;});/*** 匹配专用线程*/public static ExecutorService matchExecutorService = Executors.newSingleThreadExecutor(r -> {Thread t = new Thread(r);t.setName("RankThread");return t;});public static void init() {// 逻辑线程池logicThreadArr = new ExecutorService[Runtime.getRuntime().availableProcessors()];for (int i = 0; i < logicThreadArr.length; i++) {int finalI = i;logicThreadArr[i] = Executors.newSingleThreadExecutor(r -> {Thread t = new Thread(r);t.setName("LogicThread" + finalI);return t;});}// 房间线程池roomThreadArr = new ExecutorService[Runtime.getRuntime().availableProcessors()];for (int i = 0; i < roomThreadArr.length; i++) {int finalI = i;roomThreadArr[i] = Executors.newSingleThreadExecutor(r -> {Thread t = new Thread(r);t.setName("RoomThread" + finalI);return t;});}}
}
LogicThreadManager.java //逻辑线程池
package org.example.testLogicAndRank;import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;public class LogicThreadManager {/*** 从其他线程执行一个任务，然后将结果提交到逻辑线程** @param completableFuture* @param consumer* @param <T>*/public static <T> void executeInLogicThread(CompletableFuture<T> completableFuture, Consumer<T> consumer, Object hashObj) {ExecutorService executorService = ThreadManager.logicThreadArr[Math.abs(hashObj.hashCode()) % ThreadManager.logicThreadArr.length];completableFuture.thenAcceptAsync(consumer, executorService);}
}
RankThreadManager.java
package org.example.testLogicAndRank;import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;@Slf4j
public class RankThreadManager {public static CompletableFuture<List<Integer>> getRankList() {return submitInRankThread(() -> {// 查询数据库log.info("查询排行榜");return Lists.newArrayList(1, 2, 3, 4, 5);});}/*** 在排行榜线程执行某个操作,有返回值** @param callable* @param <T>* @return*/public static <T> CompletableFuture<T> submitInRankThread(Callable<T> callable) {return CompletableFuture.supplyAsync(() -> {try {return callable.call();} catch (Exception e) {log.error("", e);}return null;}, ThreadManager.rankExecutorService);}/*** 在排行榜线程执行某个操作,无返回值*/public static void executeInRankThread(Runnable runnable) {ThreadManager.rankExecutorService.submit(runnable);}
}
Main.java
package org.example.testLogicAndRank;import lombok.extern.slf4j.Slf4j;import java.util.List;
import java.util.concurrent.CompletableFuture;@Slf4j
public class Main {public static void main(String[] args) {ThreadManager.init();// 排行榜CompletableFuture<List<Integer>> rankListFuture = RankThreadManager.getRankList();// 假设是给玩家1和2发奖LogicThreadManager.executeInLogicThread(rankListFuture, (rankList) -> {log.info("拿到排行榜数据发奖{}", rankList);}, 1);LogicThreadManager.executeInLogicThread(rankListFuture, (rankList) -> {log.info("拿到排行榜数据发奖{}", rankList);}, 2);}
}/*
17:16:39.314 [RankThread] INFO org.example.testLogicAndRank.RankManager - 查询排行榜
17:16:39.343 [LogicThread2] INFO org.example.testLogicAndRank.Main - 拿到排行榜数据发奖[1, 2, 3, 4, 5]
17:16:39.343 [LogicThread1] INFO org.example.testLogicAndRank.Main - 拿到排行榜数据发奖[1, 2, 3, 4, 5]*/
总结：
可以看出来，我们不再需要什么Promise模式了，有了CompletableFuture后，业务线程的编排和交换数据变得非常容易了！