【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后，业务线程的编排和交换数据变得非常容易了！

代码007普通

打赏 收藏 海报 链接