Multi-Agent系统的一致性问题如何解决？共识算法如何应用

publicclassRaftAgent{
    privatevolatileRaftRole role =RaftRole.FOLLOWER;
    privatevolatileint currentTerm =0;
    privatevolatileString votedFor =null;
    privatevolatileString leaderId =null;
    privatelong lastHeartbeatTime;

    privateList<RaftAgent> peers =newArrayList<>();
    privateList<LogEntry> log =newArrayList<>();
    privateint commitIndex =0;
    privateint lastApplied =0;

    // 心跳超时检测
    privatevoidcheckHeartbeatTimeout(){
        long elapsed =System.currentTimeMillis()- lastHeartbeatTime;
        // 添加随机化避免多个节点同时发起选举
        int electionTimeout =150+newRandom().nextInt(150);// 150-300ms

        if(elapsed > electionTimeout && role ==RaftRole.FOLLOWER){
            startElection();
        }
    }

    privatevoidstartElection(){
        currentTerm++;
        role =RaftRole.CANDIDATE;
        votedFor =this.nodeId;
        int votesReceived =1;

        // 向所有其他节点请求投票
        for(RaftAgent peer : peers){
            VoteResponse response = peer.requestVote(
                currentTerm,
                nodeId,
                log.size()-1,// lastLogIndex
                log.isEmpty()?0: log.get(log.size()-1).term // lastLogTerm
            );
            if(response.isGranted()){
                votesReceived++;
            }
        }

        // 获得多数票则成为Leader
        if(votesReceived >(peers.size()+1)/2){
            becomeLeader();
        }
    }

    publicVoteResponserequestVote(int term,String candidateId,
                                  int lastLogIndex,int lastLogTerm){
        // 如果候选人的任期小于当前任期，拒绝投票
        if(term < currentTerm){
            returnnewVoteResponse(currentTerm,false);
        }

        // 如果候选人的任期更大，更新自己的任期
        if(term > currentTerm){
            currentTerm = term;
            votedFor =null;
            role =RaftRole.FOLLOWER;
        }

        // 检查日志是否至少和自己一样新
        int myLastLogIndex = log.size()-1;
        int myLastLogTerm = log.isEmpty()?0: log.get(myLastLogIndex).term;

        boolean logIsUpToDate =(lastLogTerm > myLastLogTerm)||
        (lastLogTerm == myLastLogTerm &&
         lastLogIndex >= myLastLogIndex);

        // 如果还没投票且候选人日志足够新，投票给他
        if((votedFor ==null|| votedFor.equals(candidateId))&& logIsUpToDate){
            votedFor = candidateId;
            lastHeartbeatTime =System.currentTimeMillis();
            returnnewVoteResponse(currentTerm,true);
        }

        returnnewVoteResponse(currentTerm,false);
    }

    privatevoidbecomeLeader(){
        role =RaftRole.LEADER;
        leaderId =this.nodeId;

        // 开始定期发送心跳
        startHeartbeat();
    }
}

classLogEntry{
    int term;
    String command;

    publicLogEntry(int term,String command){
        this.term = term;
        this.command = command;
    }
}

enumRaftRole{
FOLLOWER, CANDIDATE, LEADER
}

classVoteResponse{
    int term;
    boolean granted;

    publicVoteResponse(int term,boolean granted){
        this.term = term;
        this.granted = granted;
    }

    publicbooleanisGranted(){
        return granted;
    }
}

publicclassRaftAgent{
    // ... 前面的代码 ...

    // Leader处理客户端写请求
    publicbooleanhandleClientRequest(String command){
        if(role !=RaftRole.LEADER){
            returnfalse;
        }

        // 追加到本地日志
        LogEntry entry =newLogEntry(currentTerm, command);
        log.add(entry);

        // 复制到所有Follower
        int successCount =1;// 自己算一个
        for(RaftAgent peer : peers){
            AppendEntriesResponse response = peer.appendEntries(
                currentTerm,
                nodeId,
                log.size()-2,// prevLogIndex
                log.size()>=2? log.get(log.size()-2).term :0,// prevLogTerm
                Collections.singletonList(entry),
                commitIndex
            );

            if(response.isSuccess()){
                successCount++;
            }
        }

        // 如果多数节点确认，提交日志
        if(successCount >(peers.size()+1)/2){
            commitIndex = log.size()-1;
            applyToStateMachine(entry);
            returntrue;
        }

        returnfalse;
    }

    publicAppendEntriesResponseappendEntries(int term,String leaderId,
                                             int prevLogIndex,int prevLogTerm,
                                             List<LogEntry> entries,int leaderCommit){
        // 更新心跳时间
        lastHeartbeatTime =System.currentTimeMillis();

        // 任期检查
        if(term < currentTerm){
            returnnewAppendEntriesResponse(currentTerm,false);
        }

        if(term > currentTerm){
            currentTerm = term;
            votedFor =null;
        }

        role =RaftRole.FOLLOWER;
        this.leaderId = leaderId;

        // 日志一致性检查
        if(prevLogIndex >=0){
            if(prevLogIndex >= log.size()||
               log.get(prevLogIndex).term != prevLogTerm){
                returnnewAppendEntriesResponse(currentTerm,false);
            }
        }

        // 追加新日志条目
        if(entries !=null&&!entries.isEmpty()){
            int index = prevLogIndex +1;
            for(LogEntry entry : entries){
                if(index < log.size()){
                    // 如果存在冲突，删除这个索引之后的所有日志
                    if(log.get(index).term != entry.term){
                        log =newArrayList<>(log.subList(0, index));
                        log.add(entry);
                    }
                }else{
                    log.add(entry);
                }
                index++;
            }
        }

        // 更新commitIndex
        if(leaderCommit > commitIndex){
            commitIndex =Math.min(leaderCommit, log.size()-1);
            // 应用已提交的日志到状态机
            while(lastApplied < commitIndex){
                lastApplied++;
                applyToStateMachine(log.get(lastApplied));
            }
        }

        returnnewAppendEntriesResponse(currentTerm,true);
    }

    privatevoidapplyToStateMachine(LogEntry entry){
        // 将日志条目应用到状态机
        System.out.println("Applying command: "+ entry.command);
    }
}

classAppendEntriesResponse{
    int term;
    boolean success;

    publicAppendEntriesResponse(int term,boolean success){
        this.term = term;
        this.success = success;
    }

    publicbooleanisSuccess(){
        return success;
    }
}

// 预投票阶段
privatebooleanpreVote(){
    int preVoteCount =1;
    int nextTerm = currentTerm +1;

    for(RaftAgent peer : peers){
        PreVoteResponse resp = peer.requestPreVote(
            nextTerm,
            nodeId,
            log.size()-1,
            log.isEmpty()?0: log.get(log.size()-1).term
        );
        if(resp.isGranted()){
            preVoteCount++;
        }
    }
    return preVoteCount >(peers.size()+1)/2;
}

// 只有预投票通过才发起真正的选举
privatevoidcheckAndStartElection(){
    if(preVote()){
        startElection();
    }else{
        // 预投票失败，重置超时等待下次尝试
        lastHeartbeatTime =System.currentTimeMillis();
    }
}

classPreVoteResponse{
    int term;
    boolean granted;

    publicPreVoteResponse(int term,boolean granted){
        this.term = term;
        this.granted = granted;
    }

    publicbooleanisGranted(){
        return granted;
    }
}