[大数据]生产者(发出syncAsk请求)和消费者(有xxxAsk的监听，里面有实现)

home服

application.xml

 <net:config id="tankHome" protocol-location="protocol.xml" generate-cs-protocol="${net.generate.cs.protocol}"
                fold-protocol="${net.fold.protocol}">

        <net:registry center="${registry.center}" user="${registry.user}" password="${registry.password}">
            <net:address name="${registry.address.name}" url="${registry.address.url}"/>
        </net:registry>

        <net:provider task-dispatch="consistent-hash">
            <net:module name="tankHome"/>
        </net:provider>

        <net:consumer load-balancer="consistent-hash">
            <net:module name="tankCache"/>
        </net:consumer>

    </net:config>

home服发出请求

        // 异步发送给cache服
        NetContext.getConsumer().asyncAsk(BattleScoreAsk.valueOf(uid, score), BattleScoreAnswer.class, uid)
                .whenComplete(answer -> {
                    logger.info("c[{}][{}]玩家战斗结果异步回调", gatewayAttachment.getUid(), gatewayAttachment.getSid());

                    // 战斗过后如果上了排行榜，则奖励一下，每一分值一个金币，半个钻石
                    if (answer.isRankReward()) {
                        var currencyPo = player.getCurrencyPo();
                        currencyPo.setGold(currencyPo.getGold() + score);
                        currencyPo.setGem(currencyPo.getGem() + score / 2);
                        addPlayerExp(player, score);

                        playerEntityCaches.update(player);

                        NetContext.getRouter().send(session, BattleResultResponse.valueOf(score), gatewayAttachment);
                        NetContext.getRouter().send(session, CurrencyUpdateNotice.valueOf(currencyPo.toCurrencyVO()), gatewayAttachment);
                    }
                });

cache服

    @PacketReceiver
    public void atBattleScoreAsk(Session session, BattleScoreAsk ask) {
        var playerId = ask.getPlayerId();
        var score = ask.getScore();

        // 如果没有上榜，直接返回；上榜，则将当前成绩插入数据库
        if (score <= minScore && rankLimit >= RANK_SIZE) {
            NetContext.getRouter().send(session, BattleScoreAnswer.valueOf(false));
            return;
        }

        // 获取一个分布式自增唯一id
        var id = MongoIdUtils.getIncrementIdFromMongoDefault(ScoreRankEntity.class);
        // 插入数据库
        OrmContext.getAccessor().insert(ScoreRankEntity.valueOf(id, playerId, TimeUtils.now(), score));
        NetContext.getRouter().send(session, BattleScoreAnswer.valueOf(true));
    }