Unreal MassAI的网络同步基础

NoiseFloor · 发表于 2023-1-16 15:37

在了解了ECS以及Unreal中MassAI的基本使用后，我们开始考虑怎样让Entity支持网络同步，接下来我们将使用CitySample中的Crowd作为示例进行讲解
UMassReplicationTrait

简单来说，Entity的网络同步只需为其添加MassReplicationTrait，然后再指定用于服务端与客户端通信所用的TSubclassOf<AMassClientBubbleInfoBase>以及用于在服务器端处理Entity同步的TSubclassOf<UMassReplicatorBase>即可。下面我们先来具体的看一下MassReplicationTrait相关的配置
有了ReplicationTrait，并进行了正确的配置后CrowdEntity就能够支持网络同步了。
接下来如果我们有一种新类型的Entity要支持网络同步，我们要为此做些什么？Unreal的Mass AI具体又是怎样实现的网络同步？
下面我们先看看UMassReplicationTrait代码
void UMassReplicationTrait::BuildTemplate(FMassEntityTemplateBuildContext& BuildContext, const UWorld& World) const
{
if (World.IsNetMode(NM_Standalone))
{
      return;
}

FReplicationTemplateIDFragment& TemplateIDFragment = BuildContext.AddFragment_GetRef<FReplicationTemplateIDFragment>();
TemplateIDFragment.ID = BuildContext.GetTemplateID();

BuildContext.AddFragment<FMassNetworkIDFragment>();
BuildContext.AddFragment<FMassReplicatedAgentFragment>();
BuildContext.AddFragment<FMassReplicationViewerInfoFragment>();
BuildContext.AddFragment<FMassReplicationLODFragment>();
BuildContext.AddFragment<FMassReplicationGridCellLocationFragment>();

FMassEntityManager& EntityManager = UE::Mass::Utils::GetEntityManagerChecked(World);

UMassReplicationSubsystem* ReplicationSubsystem = UWorld::GetSubsystem<UMassReplicationSubsystem>(&World);
check(ReplicationSubsystem);

FConstSharedStruct ParamsFragment = EntityManager.GetOrCreateConstSharedFragment(Params);
BuildContext.AddConstSharedFragment(ParamsFragment);

uint32 ParamsHash = UE::StructUtils::GetStructCrc32(FConstStructView::Make(Params));
FSharedStruct SharedFragment = EntityManager.GetOrCreateSharedFragmentByHash<FMassReplicationSharedFragment>(ParamsHash, *ReplicationSubsystem, Params);
BuildContext.AddSharedFragment(SharedFragment);
}
根据上述的一些信息我们先来个纸上谈兵：我们首先要知道服务端会拥有所有要同步的Entity，通常客户端会有的是上述的一个子集。那么如何确保服务端与客户端Entity的&#34;一致性&#34;，本质上是要让Entity拥有相同的Fragments，并且这些Fragments中的数据也相同即可。要让服务端与客户端的Entity类型一致，需要添加并同步FReplicationTemplateIDFragment 要让服务端与客户端的Entity实例一致，需要添加并同步FMassNetworkIDFragment
服务端通常不会（也没有必要）将所有Entity同步到客户端，这里就会出现一个问题，服务端如何决定将哪些Entity同步到哪个客户端呢？在服务端会通过FMassReplicationViewerInfoFragment记录每个客户端（Viewer）与Entity的距离，然后可以通过距离过滤。这里需要注意FMassReplicationViewerInfoFragment虽然在客户端也存在（应该是用于保持Entity类型的一致性）但客户端并不会使用这个数据，FMassReplicationViewerInfoFragment仅在UMassReplicationProcessor中使用，UMassReplicationProcessor只在服务端执行ExecutionFlags = int32(EProcessorExecutionFlags::Server)。
AMassClientBubbleInfoBase

接下来我们顺藤摸瓜，先从AMassClientBubbleInfoBase开始分析。
首先要考虑怎样在服务端和客户端之间进行Entity和Fragment的同步呢？
/** The info actor base class that provides the actual replication */
UCLASS()
class MASSREPLICATION_API AMassClientBubbleInfoBase : public AInfo
{
TArray<FMassClientBubbleSerializerBase*> Serializers;
};

AMassClientBubbleInfoBase::AMassClientBubbleInfoBase(const FObjectInitializer& ObjectInitializer)
: Super(ObjectInitializer)
{
bReplicates = true;
bOnlyRelevantToOwner = true;
bNetUseOwnerRelevancy = true;
}
Entity和Fragment并不是直接进行Replication，而是通过一个支持Replication的Actor和FastArray进行网络同步。这个Actor叫AMassClientBubbleInfoBase（眼熟吧），通过上述代码可以了解到这个Actor会在服务端和所属的客户端之间进行同步。在UMassReplicationProcessor中会通过UMassReplicationSubsystem为Client创建这个Actor，并设置其Owner为Client的PlayerController。
至此服务端和每个客户端之间的通讯就有了一个载体，接下来要考虑如何进行Entity创建或删除以及Fragment数据的同步，答案是AMassClientBubbleInfoBase中的TArray<FMassClientBubbleSerializerBase*> Serializers。不啰嗦直接贴Crowd中的代码
UCLASS()
class MASSCROWD_API AMassCrowdClientBubbleInfo : public AMassClientBubbleInfoBase
{
protected:
virtual void GetLifetimeReplicatedProps(TArray<FLifetimeProperty>& OutLifetimeProps) const override;

protected:
UPROPERTY(Replicated, Transient)
FMassCrowdClientBubbleSerializer CrowdSerializer;
};
很显然FMassCrowdClientBubbleSerializer是FMassClientBubbleSerializerBase的派生，并且CrowdSerializer标识了Replicated。下面我们要搞清楚FMassCrowdClientBubbleSerializer中到底有哪些数据，服务端如何维护这个数据，客户端收到数据更新后又要如何处理。
FMassClientBubbleSerializerBase

下面先来简单的认识一下FMassClientBubbleSerializerBase
USTRUCT()
struct MASSREPLICATION_API FMassClientBubbleSerializerBase : public FFastArraySerializer
{
GENERATED_BODY()

public:

#if UE_REPLICATION_COMPILE_CLIENT_CODE
void PreReplicatedRemove(const TArrayView<int32> RemovedIndices, int32 FinalSize) const;
void PostReplicatedAdd(const TArrayView<int32> AddedIndices, int32 FinalSize) const;
void PostReplicatedChange(const TArrayView<int32> ChangedIndices, int32 FinalSize) const;
#endif //UE_REPLICATION_COMPILE_CLIENT_CODE

/** Pointer to the IClientBubbleHandlerInterface derived class in the class derived from this one */
IClientBubbleHandlerInterface* ClientHandler = nullptr;
};
这里需要先了解FFastArraySerializer是如何使用的，简单来说在FFastArraySerializer的派生中会有一个TArray<FMyData>然后通过Traits的NetDeltaSerializer方法对TArray<FMyData>中的数据进行增量式的网络同步，当有数据添加或删除时会在客户端执行PostReplicatedAdd或PreReplicatedRemove，有数据修改时会在客户端执行PostReplicatedChange。更多细节请自行Google^^
在上述代码中还有一个非常重要的接口指针IClientBubbleHandlerInterface* ClientHandler = nullptr;，ClientHandler的本质是职责的下放，将要同步的Entity数据的增删改下放到了此接口中，我们稍后再看IClientBubbleHandlerInterface的实现。
接下来我们先回到Serializer，先来看一下最终版的FMassCrowdClientBubbleSerializer
USTRUCT()
struct MASSCROWD_API FMassCrowdClientBubbleSerializer : public FMassClientBubbleSerializerBase
{
GENERATED_BODY()

FMassCrowdClientBubbleSerializer()
{
      Bubble.Initialize(Crowd, *this);
};

bool NetDeltaSerialize(FNetDeltaSerializeInfo& DeltaParams)
{
      return FFastArraySerializer::FastArrayDeltaSerialize<FCrowdFastArrayItem, FMassCrowdClientBubbleSerializer>(Crowd, DeltaParams, *this);
}

public:
FMassCrowdClientBubbleHandler Bubble;

protected:
UPROPERTY(Transient)
TArray<FCrowdFastArrayItem> Crowd;
};
到这里我们终于看到了实际要同步的数据TArray<FCrowdFastArrayItem> Crowd，要特别注意这里并不需要为UPROPERTY标识Replicated，而且FCrowdFastArrayItem里的数据成员也不需要，这自然归功于NetDeltaSerialize和U++中的反射。FMassCrowdClientBubbleSerializer中还包含了FMassCrowdClientBubbleHandler Bubble;这是IClientBubbleHandlerInterface接口的一个实现，我们暂且放过她。
FMassFastArrayItemBase

下面我们直接贴代码，先看看FCrowdFastArrayItem到底是个什么玩意
USTRUCT()
struct FMassFastArrayItemBase : public FFastArraySerializerItem

USTRUCT()
struct MASSCROWD_API FCrowdFastArrayItem : public FMassFastArrayItemBase
{
UPROPERTY()
FReplicatedCrowdAgent Agent;
};
到这里我们会发现FCrowdFastArrayItem本质是一个FFastArraySerializerItem。有没有注意到同步的数据结构中目前为止并没有Entity或Fragment的踪影，反而出现了一个叫ReplicatedXXXAgent的玩意
FReplicatedAgentBase

继续贴Agent的代码
USTRUCT()
struct FReplicatedAgentBase
{
private:
UPROPERTY()
FMassNetworkID NetID; //uint32 Value;

UPROPERTY()
FMassEntityTemplateID TemplateID; // uint32 Hash;EMassEntityTemplateIDType Type;
};

USTRUCT()
struct MASSCROWD_API FReplicatedCrowdAgent : public FReplicatedAgentBase
{
const FReplicatedAgentPathData& GetReplicatedPathData() const { return Path; }
FReplicatedAgentPathData& GetReplicatedPathDataMutable() { return Path; }

const FReplicatedAgentPositionYawData& GetReplicatedPositionYawData() const { return PositionYaw; }
FReplicatedAgentPositionYawData& GetReplicatedPositionYawDataMutable() { return PositionYaw; }

private:
UPROPERTY(Transient)
FReplicatedAgentPathData Path;

UPROPERTY(Transient)
FReplicatedAgentPositionYawData PositionYaw;
};

USTRUCT()
struct MASSREPLICATION_API FReplicatedAgentPositionYawData
{
private:
UPROPERTY(Transient)
FVector Position;

/** Yaw in radians */
UPROPERTY(Transient)
float Yaw = 0;
};
虽然上面还是不见Entity和Fragment，但通过Agent里面的内容我们也不难猜到一个Agent&#34;就是&#34;一个Entity。MassEntity的NetID和TemplateID，CrowdEntity的Position和Yaw ^^ 到此为止整个Entity要同步的数据结构全部了然，这里省略了FReplicatedAgentPathData其内部主要是用于ZoneGraph寻路相关的数据，有需要可以直接读源码这里不再赘述。总结下，一个FCrowdFastArrayItem就代表了一个CrowdEntity，那么整个Crowd自然就代表了要同步给这个客户端的所有CrowdEntity。
IClientBubbleHandlerInterface

下面来到了重头戏，TArray<FCrowdFastArrayItem> Crowd中的数据在服务端应该如何维护，客户端收到数据变化后又该如何处理呢？让我们想起曾经的她，还记得FMassCrowdClientBubbleSerializer中除了有Crowd还有一个FMassCrowdClientBubbleHandler Bubble 吗 FMassCrowdClientBubbleHandler本质是IClientBubbleHandlerInterface接口的一个实现，我们先来看接口
class IClientBubbleHandlerInterface
{
public:
virtual ~IClientBubbleHandlerInterface() {}

virtual void InitializeForWorld(UWorld& InWorld) = 0;

#if UE_REPLICATION_COMPILE_CLIENT_CODE
// 这里可以回忆一下FastArray
virtual void PreReplicatedRemove(const TArrayView<int32> RemovedIndices, int32 FinalSize) = 0;
virtual void PostReplicatedAdd(const TArrayView<int32> AddedIndices, int32 FinalSize) = 0;
virtual void PostReplicatedChange(const TArrayView<int32> ChangedIndices, int32 FinalSize) = 0;
#endif // UE_REPLICATION_COMPILE_CLIENT_CODE

virtual void Reset() = 0; // 重置
virtual void UpdateAgentsToRemove() = 0; // 看字面意思是更新要删除的Agents

virtual void Tick(float DeltaTime) = 0;
virtual void SetClientHandle(FMassClientHandle InClientHandle) = 0; // 看来每个ClientBubbleHandler都会有个MassClientHandle

virtual void DebugValidateBubbleOnServer() = 0;
virtual void DebugValidateBubbleOnClient() = 0;
};

// 在服务端上的客户端句柄. 每个句柄可理解为带有PlayerController的NetConnection
USTRUCT()
struct MASSREPLICATION_API FMassClientHandle : public FIndexedHandleBase
上述代码中的UE_REPLICATION_COMPILE_CLIENT_CODE宏标识了其中的代码只在客户端存在和执行，其中有三个方法PreReplicatedRemove、PostReplicatedAdd、PostReplicatedChange这里可以回忆一下FastArray^^ 通过SetClientHandle方法可以看出来ClientBubbleHandler中还会有个MassClientHandle，Handler的含义是某种/某些事件的处理器，Handle是某个类型对象的句柄。我们知道ClientBubbleHandler代表了要处理某个客户端的(Entity -> Agent) on server -> NetDeltaSerialize -> (Agent -> Entity) on clien，那么具体是哪一个客户端就由SetClientHandle方法来指定。另外要特别注意此方法只在服务端执行。
TClientBubbleHandlerBase

对于不管是Crowd还是Vehicle等具体的Entity的同步过程，在服务端和客户端都有一些泛化的事情要处理，所以就有了IClientBubbleHandlerInterface接口的泛化实现TClientBubbleHandlerBase，如果我们有一种新类型的Entity要进行网络同步直接从TClientBubbleHandlerBase派生即可。那么在TClientBubbleHandlerBase中具体做了哪些事情，我们直接帖代码，同时对于TClientBubbleHandlerBase中成员的作用我们直接在代码中通过注释说明
template<typename AgentArrayItem>
class TClientBubbleHandlerBase : public IClientBubbleHandlerInterface
{
public:
/** 初始化，在服务端和客户端都会执行 */
virtual void Initialize(TArray<AgentArrayItem>& InAgents, FMassClientBubbleSerializerBase& InSerializer);
virtual void InitializeForWorld(UWorld& InWorld) override;

#if UE_REPLICATION_COMPILE_SERVER_CODE // 仅在服务端存在和执行
// 这里的几个方法本质是在服务端根据要同步的Entities维护Agents，不赘述
FMassReplicatedAgentHandle AddAgent(FMassEntityHandle Entity, typename AgentArrayItem::FReplicatedAgentType& Agent);

bool RemoveAgent(FMassNetworkID NetID);
bool RemoveAgent(FMassReplicatedAgentHandle AgentHandle);
void RemoveAgentChecked(FMassReplicatedAgentHandle AgentHandle);

/** Gets an agent safely */
const typename AgentArrayItem::FReplicatedAgentType* GetAgent(FMassReplicatedAgentHandle Handle) const;

/** Faster version to get an agent that performs check()s for debugging */
const typename AgentArrayItem::FReplicatedAgentType& GetAgentChecked(FMassReplicatedAgentHandle Handle) const;
const TArray<AgentArrayItem>& GetAgents() const { return *Agents; }
#endif //UE_REPLICATION_COMPILE_SERVER_CODE

protected:
#if UE_REPLICATION_COMPILE_CLIENT_CODE // 仅在客户端存在和执行
// 即将要删除某些Agents，先删除对应的Entity
virtual void PreReplicatedRemove(const TArrayView<int32> RemovedIndices, int32 FinalSize) override;

/** 添加某些Agent后，用于更新或创建Agent对应的Entity的辅助方法 */
/** 添加Agent后，并不一定要创建新的Entity，也可能会使用已经存在的Entity，这样只需要通过SetModifiedEntityData函数指针同步Agent和Entity的数据即可 */
/** 如果需要创建新的Entity，就交给PostReplicatedAddEntitiesHelper处理，这时候需要使用SetSpawnedEntityData函数指针同步Agent和Entity的数据 */
/** 需要在派生类的PostReplicatedAdd中调用 */
void PostReplicatedAddHelper(const TArrayView<int32> AddedIndices, FAddRequirementsForSpawnQueryFunction AddRequirementsForSpawnQuery
      , FCacheFragmentViewsForSpawnQueryFunction CacheFragmentViewsForSpawnQuery, FSetSpawnedEntityDataFunction SetSpawnedEntityData, FSetModifiedEntityDataFunction SetModifiedEntityData);
void PostReplicatedAddEntitiesHelper(const TArrayView<int32> AddedIndices, FAddRequirementsForSpawnQueryFunction AddRequirementsForSpawnQuery
      , FCacheFragmentViewsForSpawnQueryFunction CacheFragmentViewsForSpawnQuery, FSetSpawnedEntityDataFunction SetSpawnedEntityData);

/** 某些Agent的数据发生变化时通过SetModifiedEntityData函数指针同步其对应的Entity数据，需要在派生类的PostReplicatedChange中调用 */
void PostReplicatedChangeHelper(const TArrayView<int32> ChangedIndices, FSetModifiedEntityDataFunction SetModifiedEntityData);
#endif //UE_REPLICATION_COMPILE_SERVER_CODE

protected:
FMassClientBubbleSerializerBase* Serializer = nullptr; // 我是个工具人，Serializer是我的老板，我工作的进展也会即时的通知到老板(MarkItemDirty)
TArray<AgentArrayItem>* Agents = nullptr; // 把工作的结果直接同步给老板，注意这里的Agents是个指针，其真正存在于上面的Serializer中

// 我是个工具人，我为哪个客户工作，有了这个才能知道客户需要同步哪些Entity（这件事情是在UMassReplicationProcessor中处理）
// 注意，ClientHandle只在服务端有效，其会在UMassReplicationProcessor::PrepareExecution方法中进行设置，即使这里目前并没有用到这个ClientHandle
FMassClientHandle ClientHandle;

// 下面是为了完成上述各种功能的一些辅助型数据成员，不再赘述
#if UE_REPLICATION_COMPILE_SERVER_CODE
FMassReplicatedAgentHandleManager AgentHandleManager;
TArray<FMassAgentLookupData> AgentLookupArray;
TMap<FMassNetworkID, FMassReplicatedAgentHandle> NetworkIDToAgentHandleMap;
#endif //UE_REPLICATION_COMPILE_SERVER_CODE

#if UE_REPLICATION_COMPILE_CLIENT_CODE
TMap<FMassNetworkID, FMassAgentRemoveData> AgentsRemoveDataMap;
#endif //UE_REPLICATION_COMPILE_CLIENT_CODE
};
到目前为止我们已经可以完成Entity创建和删除的同步，那么Entity中的数据也就是Fragments是如何进行同步的呢？再具体些：在服务端怎样把Entity的Fragments数据复制到Agent？在客户端怎样把Agent中的数据复制到Entity的Fragments？
FMassCrowdClientBubbleHandler

我们将在ClientBubbleHandler的最终版FMassCrowdClientBubbleHandler中一窥究竟
class MASSCROWD_API FMassCrowdClientBubbleHandler : public TClientBubbleHandlerBase<FCrowdFastArrayItem>
{
public:
typedef TMassClientBubblePathHandler<FCrowdFastArrayItem> FMassClientBubblePathHandler;
typedef TMassClientBubbleTransformHandler<FCrowdFastArrayItem> FMassClientBubbleTransformHandler;

FMassCrowdClientBubbleHandler()
      : PathHandler(*this)
      , TransformHandler(*this)
{}

#if UE_REPLICATION_COMPILE_SERVER_CODE
const FMassClientBubblePathHandler& GetPathHandler() const { return PathHandler; }
FMassClientBubblePathHandler& GetPathHandlerMutable() { return PathHandler; }

const FMassClientBubbleTransformHandler& GetTransformHandler() const { return TransformHandler; }
FMassClientBubbleTransformHandler& GetTransformHandlerMutable() { return TransformHandler; }
#endif // UE_REPLICATION_COMPILE_SERVER_CODE

protected:
#if UE_REPLICATION_COMPILE_CLIENT_CODE
virtual void PostReplicatedAdd(const TArrayView<int32> AddedIndices, int32 FinalSize) override;
virtual void PostReplicatedChange(const TArrayView<int32> ChangedIndices, int32 FinalSize) override;

void PostReplicatedChangeEntity(const FMassEntityView& EntityView, const FReplicatedCrowdAgent& Item) const;
#endif //UE_REPLICATION_COMPILE_CLIENT_CODE

FMassClientBubblePathHandler PathHandler;
FMassClientBubbleTransformHandler TransformHandler;
};
简单回顾一下，FCrowdFastArrayItem里面有个Agent，Agent里面有PathData和PositionYawData。在FMassCrowdClientBubbleHandler中会分别使用FMassClientBubblePathHandler和FMassClientBubbleTransformHandler完成数据的Copy，在服务端是Entity->Agent，在客户端是Agent->Entity。 PathHandler对应PathData，TransformHandler对应PoisitionYawData，我们稍后再看这两个Handler是如何完成工作的。
下面让我们先从客户端的角度来看看当有新的FCrowdFastArrayItem从服务端同步过来时应该如何处理
#if UE_REPLICATION_COMPILE_CLIENT_CODE
void FMassCrowdClientBubbleHandler::PostReplicatedAdd(const TArrayView<int32> AddedIndices, int32 FinalSize)
{
auto AddRequirementsForSpawnQuery = [this](FMassEntityQuery& InQuery)
{
      PathHandler.AddRequirementsForSpawnQuery(InQuery);
      TransformHandler.AddRequirementsForSpawnQuery(InQuery);
};

auto CacheFragmentViewsForSpawnQuery = [this](FMassExecutionContext& InExecContext)
{
      PathHandler.CacheFragmentViewsForSpawnQuery(InExecContext);
      TransformHandler.CacheFragmentViewsForSpawnQuery(InExecContext);
};

auto SetSpawnedEntityData = [this](const FMassEntityView& EntityView, const FReplicatedCrowdAgent& ReplicatedEntity, const int32 EntityIdx)
{
      PathHandler.SetSpawnedEntityData(EntityView, ReplicatedEntity.GetReplicatedPathData(), EntityIdx);
      TransformHandler.SetSpawnedEntityData(EntityIdx, ReplicatedEntity.GetReplicatedPositionYawData());
};

auto SetModifiedEntityData = [this](const FMassEntityView& EntityView, const FReplicatedCrowdAgent& Item)
{
      PostReplicatedChangeEntity(EntityView, Item);
};

PostReplicatedAddHelper(AddedIndices, AddRequirementsForSpawnQuery, CacheFragmentViewsForSpawnQuery, SetSpawnedEntityData, SetModifiedEntityData);

PathHandler.ClearFragmentViewsForSpawnQuery();
TransformHandler.ClearFragmentViewsForSpawnQuery();
}
#endif //UE_REPLICATION_COMPILE_SERVER_CODE
看到这里是否了然^^，你可曾记得FMassCrowdClientBubbleHandler的父类TClientBubbleHandlerBase中有个PostReplicatedAddHelper方法？此方法会为我们根据新的Agent创建其对应的Entity，这个新的Entity中的Fragments的数据将由这里的匿名函数SetSpawnedEntityData通过DataHandler完成从Agent到Entity的Copy，其他匿名函数不再赘述。同样的在客户端删除和更新Entity的过程也不再赘述，可自行查阅相关代码。
UMassReplicatorBase

接下来让我们来到服务端，怎样把Entity同步到Agent？这里再贴一下FMassCrowdClientBubbleHandler中与Server相关的代码
class MASSCROWD_API FMassCrowdClientBubbleHandler : public TClientBubbleHandlerBase<FCrowdFastArrayItem>
{
public:
#if UE_REPLICATION_COMPILE_SERVER_CODE
const FMassClientBubblePathHandler& GetPathHandler() const { return PathHandler; }
FMassClientBubblePathHandler& GetPathHandlerMutable() { return PathHandler; }

const FMassClientBubbleTransformHandler& GetTransformHandler() const { return TransformHandler; }
FMassClientBubbleTransformHandler& GetTransformHandlerMutable() { return TransformHandler; }
#endif // UE_REPLICATION_COMPILE_SERVER_CODE
}
这里只有Handler对应的Getter方法，下面在查找对GetPathHandlerMutable的调用之后，我们就来到了一个新的类中UMassCrowdReplicator。
UCLASS()
class MASSCROWD_API UMassCrowdReplicator : public UMassReplicatorBase
{
GENERATED_BODY()

public:
void AddRequirements(FMassEntityQuery& EntityQuery) override;
void ProcessClientReplication(FMassExecutionContext& Context, FMassReplicationContext& ReplicationContext) override;
};
UMassReplicatorBase，对她是否还有印象？还记得ReplicationTrait的Params里有个TSubclassOf<UMassReplicatorBase> ReplicatorClass吗？最初的“简单来说...” 终于要集齐&#34;七颗龙珠&#34;了
在服务端将Entity同步到Agent的过程就由这个UMassReplicatorBase来完成，而且这个UMassReplicatorBase仅在服务端的UMassReplicationProcessor中使用下面我们来看UMassCrowdReplicator::ProcessClientReplication方法怎样将Entity的Fragments中的数据复制到Agent中的
void UMassCrowdReplicator::ProcessClientReplication(FMassExecutionContext& Context, FMassReplicationContext& ReplicationContext)
{
#if UE_REPLICATION_COMPILE_SERVER_CODE
FMassReplicationProcessorPathHandler PathHandler;
FMassReplicationProcessorPositionYawHandler PositionYawHandler;
FMassReplicationSharedFragment* RepSharedFrag = nullptr;

auto CacheViewsCallback = [&RepSharedFrag, &PathHandler, &PositionYawHandler](FMassExecutionContext& Context)
{
      PathHandler.CacheFragmentViews(Context);
      PositionYawHandler.CacheFragmentViews(Context);
      RepSharedFrag = &Context.GetMutableSharedFragment<FMassReplicationSharedFragment>();
      check(RepSharedFrag);
};

auto AddEntityCallback = [&RepSharedFrag, &PathHandler, &PositionYawHandler](FMassExecutionContext& Context,const int32 EntityIdx, FReplicatedCrowdAgent& InReplicatedAgent, const FMassClientHandle ClientHandle)->FMassReplicatedAgentHandle
{
      AMassCrowdClientBubbleInfo& CrowdBubbleInfo = RepSharedFrag->GetTypedClientBubbleInfoChecked<AMassCrowdClientBubbleInfo>(ClientHandle);

      PathHandler.AddEntity(EntityIdx, InReplicatedAgent.GetReplicatedPathDataMutable());
      PositionYawHandler.AddEntity(EntityIdx, InReplicatedAgent.GetReplicatedPositionYawDataMutable());

      return CrowdBubbleInfo.GetCrowdSerializer().Bubble.AddAgent(Context.GetEntity(EntityIdx), InReplicatedAgent);
};

auto ModifyEntityCallback = [&RepSharedFrag, &PathHandler](FMassExecutionContext& Context, const int32 EntityIdx, const EMassLOD::Type LOD, const float Time, const FMassReplicatedAgentHandle Handle, const FMassClientHandle ClientHandle)
{
      AMassCrowdClientBubbleInfo& CrowdBubbleInfo = RepSharedFrag->GetTypedClientBubbleInfoChecked<AMassCrowdClientBubbleInfo>(ClientHandle);

      FMassCrowdClientBubbleHandler& Bubble = CrowdBubbleInfo.GetCrowdSerializer().Bubble;

      const bool bLastClient = RepSharedFrag->CachedClientHandles.Last() == ClientHandle;
      PathHandler.ModifyEntity<FCrowdFastArrayItem>(Handle, EntityIdx, Bubble.GetPathHandlerMutable(), bLastClient);

      // Don&#39;t call the PositionYawHandler here as we currently only replicate the position and yaw when we add an entity to Mass
};

auto RemoveEntityCallback = [&RepSharedFrag](FMassExecutionContext& Context, const FMassReplicatedAgentHandle Handle, const FMassClientHandle ClientHandle)
{
      AMassCrowdClientBubbleInfo& CrowdBubbleInfo = RepSharedFrag->GetTypedClientBubbleInfoChecked<AMassCrowdClientBubbleInfo>(ClientHandle);

      CrowdBubbleInfo.GetCrowdSerializer().Bubble.RemoveAgentChecked(Handle);
};

CalculateClientReplication<FCrowdFastArrayItem>(Context, ReplicationContext, CacheViewsCallback, AddEntityCallback, ModifyEntityCallback, RemoveEntityCallback);
#endif // UE_REPLICATION_COMPILE_SERVER_CODE
}
这里又多了两个Handler，完整命名是FMassReplicationProcessor???Handler，这玩意只在服务端使用。
这里先注意FMassCrowdClientBubbleHandler上面提到的Handler完整的类型名称叫FMassClientBubble???Handler，这玩意在客户端和服务端都会使用。在下面的UMassCrowdReplicator中也会有FMassReplicationProcessor???Handler，这玩意只在服务端使用。关于这两种Handler的更多细节将在稍后说明。
UMassReplicationProcessor

最后我们来看一下在服务端用于同步Entity的大总管UMassReplicationProcessor，这里我们直接在代码中通过注释的方式进行讲解
UMassReplicationProcessor::UMassReplicationProcessor()
{
ExecutionFlags = int32(EProcessorExecutionFlags::Server);
ProcessingPhase = EMassProcessingPhase::PostPhysics;

// 由于MassReplicationProcessor要同步Clients和Viewers（UMassReplicationSubsystem::SynchronizeClientsAndViewers会调用SpawnActor），所以要限制Processor只在GameThread执行
bRequiresGameThreadExecution = true;
}

void UMassReplicationProcessor::PrepareExecution(FMassEntityManager& EntityManager)
{
#if UE_REPLICATION_COMPILE_SERVER_CODE
check(ReplicationSubsystem);

// 更新客户端和观察者的信息，一个客户端可以有多个观察者
// 同时更新后的信息也将保存在ReplicationSubsystem中
ReplicationSubsystem->SynchronizeClientsAndViewers();

// 有多少种要同步的Entity就有多少个MassReplicationSharedFragment，详见UMassReplicationTrait::BuildTemplate
EntityManager.ForEachSharedFragment<FMassReplicationSharedFragment>([this](FMassReplicationSharedFragment& RepSharedFragment)
{
      if (!RepSharedFragment.bEntityQueryInitialized)
      {
         // MassReplicationProcessor中的EntityQuery会查询出所有要同步的Entity
         RepSharedFragment.EntityQuery = EntityQuery;

         // 已知一个RepSharedFragment对应一种特定类型的Entity（每个RepSharedFragment中都有一个专属的Replicator为这种Entity处理网络同步）
         // 因此需要让RepSharedFragment.EntityQuery只返回其对应种类的Entity，这就是SetChunkFilter的作用
         RepSharedFragment.EntityQuery.SetChunkFilter([&RepSharedFragment](const FMassExecutionContext& Context)
         {
            const FMassReplicationSharedFragment& CurRepSharedFragment = Context.GetSharedFragment<FMassReplicationSharedFragment>();
            return &CurRepSharedFragment == &RepSharedFragment;
         });
         RepSharedFragment.CachedReplicator->AddRequirements(RepSharedFragment.EntityQuery);
         RepSharedFragment.bEntityQueryInitialized = true;
      }

      // 像RepSharedFragment中的Replicator一样，RepSharedFragment里也有一个专属的BubbleInfos，代表了RepSharedFragment对应类型的Entity在服务端与每个客户端的网络同步的&#34;通道&#34;
      // 下面还有一大段代码，作用是将ReplicationSubsystem中的Client信息同步给RepSharedFragment，这里不再赘述
      ......
});
#endif //UE_REPLICATION_COMPILE_SERVER_CODE
}
在UMassReplicationProcessor::PrepareExecution执行后，我们就准备好了MassReplicationSharedFragments。她们知道怎样找到各自负责的所有Entity，也知道要将这些Entity同步给哪些客户端
PrepareExecution方法会在UMassReplicationProcessor::Execute方法中调用，下面我们就在UMassReplicationProcessor::Execute中一窥具体的同步过程
void UMassReplicationProcessor::Execute(FMassEntityManager& EntityManager, FMassExecutionContext& Context)
{
#if UE_REPLICATION_COMPILE_SERVER_CODE
UWorld* World = EntityManager.GetWorld();
check(World);
check(ReplicationSubsystem);

{
      QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_Preperation);
      PrepareExecution(EntityManager);
}

const UMassLODSubsystem& LODSubsystem = Context.GetSubsystemChecked<UMassLODSubsystem>(EntityManager.GetWorld());
const TArray<FViewerInfo>& AllViewersInfo = LODSubsystem.GetViewers();
const TArray<FMassClientHandle>& ClientHandles = ReplicationSubsystem->GetClientReplicationHandles();

// Execute中同步处理的最外层是以FMassClientHandle为维度的一个大循环体，也就是说这个循环每次执行都为某个具体的客户端完成了Entity的同步
for (const FMassClientHandle ClientHandle : ClientHandles)
{
      if (ReplicationSubsystem->IsValidClientHandle(ClientHandle) == false)
      {
         continue;
      }

      // 这里需要特别注意，在网络同步相关的Fragment（比如FMassReplicationSharedFragment，FMassReplicationLODFragment）中并不会记录所有客户端同步相关的数据
      // 所有客户端同步相关的数据会被记录在ReplicationSubsystem的ClientsReplicationInfo中
      // ClientReplicationInfo代表的是当前正在处理的客户端同步所需数据
      // 下面做LOD计算时，Fragment中所需的某些数据就会从这个ClientReplicationInfo中获取的
      FMassClientReplicationInfo& ClientReplicationInfo = ReplicationSubsystem->GetMutableClientReplicationInfoChecked(ClientHandle);

      // 一个Client可能会有多个Viewer，比如一个客户端多个玩家时？
      TArray<FViewerInfo> Viewers;
      for (const FMassViewerHandle ClientViewerHandle : ClientReplicationInfo.Handles)
      {
         const FViewerInfo* ViewerInfo = AllViewersInfo.FindByPredicate([ClientViewerHandle](const FViewerInfo& ViewerInfo) { return ClientViewerHandle == ViewerInfo.Handle; });
         if (ensureMsgf(ViewerInfo, TEXT(&#34;Expecting to find the client viewer handle in the all viewers info list&#34;)))
         {
            Viewers.Add(*ViewerInfo);
         }
      }

      // 我们并不会也不需要把服务端所有要同步的Entity同步到客户端，下面主要就是处理这个问题，具体就是把Entity通过LOD（距离）进行筛选

      // 更新每种Entity对应的FMassReplicationSharedFragment的Viewers，后面会用来进行LOD相关的计算
      // 还要记录下最远的LOD距离，注意这里忽视了Entity的类型，也就是不管什么类型的Entity只要一个最远的
      float MaxLODDistance = 0.0f;
      EntityManager.ForEachSharedFragment<FMassReplicationSharedFragment>([&Viewers,&MaxLODDistance](FMassReplicationSharedFragment& RepSharedFragment)
      {
         RepSharedFragment.LODCollector.PrepareExecution(Viewers);
         RepSharedFragment.LODCalculator.PrepareExecution(Viewers);
         MaxLODDistance = FMath::Max(MaxLODDistance, RepSharedFragment.LODCalculator.GetMaxLODDistance());
      });

      // 这里会通过MaxLODDistance以及Grid进行粗略的筛选，最后只有在MaxLODDistance以内的Entity才会保留在EntitiesInRange中
      // 如果一个Client有多个Viewer，EntitiesInRange中的Entity可能会有重复
      const FVector HalfExtent(MaxLODDistance, MaxLODDistance, 0.0f);
      TArray<FMassEntityHandle> EntitiesInRange;
      for (const FViewerInfo& Viewer : Viewers)
      {
         FBox Bounds(Viewer.Location - HalfExtent, Viewer.Location + HalfExtent);
         ReplicationSubsystem->GetGrid().Query(Bounds, EntitiesInRange);
      }

      // 如果存在支持网络同步的Entity才开始真正执行网络同步
      EntityQuery.CacheArchetypes(EntityManager);
      if (EntityQuery.GetArchetypes().Num() > 0)
      {
         // 根据Entity的类型（Archetype）对Entities进行分组存储
         struct FEntitySet
         {
            void Reset()
            {
                  Entities.Reset();
            }

            FMassArchetypeHandle Archetype;
            TArray<FMassEntityHandle> Entities;
         };
         TArray<FEntitySet> EntitySets;

         for (const FMassArchetypeHandle& Archetype : EntityQuery.GetArchetypes())
         {
            FEntitySet& Set = EntitySets.AddDefaulted_GetRef();
            Set.Archetype = Archetype;
         }

         // 匿名函数，用于将Entities以分组的方式存储在EntitySets中
         // 内部实现简单优化了一下性能
         // 思路是为了在存储时避免每次都要在EntitySets中查找类型匹配的Set，先假设下一个Entity的类型和当前的Entity类型是一样的，如果是就不用找了，如果不是再找
         auto BuildEntitySet = [&EntitySets, &EntityManager](const TArray<FMassEntityHandle>& Entities)
         {
            FEntitySet* PrevSet = Entities.Num() ? &EntitySets[0] : nullptr;
            for (const FMassEntityHandle Entity : Entities)
            {
                  // Add to set of supported archetypes. Dont process if we don&#39;t care about the type.
                  const FMassArchetypeHandle Archetype = EntityManager.GetArchetypeForEntity(Entity);
                  FEntitySet* Set = PrevSet && PrevSet->Archetype == Archetype ? PrevSet : EntitySets.FindByPredicate([&Archetype](const FEntitySet& Set) { return Archetype == Set.Archetype; });
                  if (Set != nullptr)
                  {
                     // We don&#39;t care about duplicates here, the FMassArchetypeEntityCollection creation below will handle it
                     Set->Entities.Add(Entity);
                     PrevSet = Set;
                  }
            }
         };

         // 看这个方法的最后可以知道ClientReplicationInfo.HandledEntities中存放的就是上一次的EntitiesInRange
         // 从代码执行逻辑上来说上一次的EntitiesInRange即便在这次没有被筛选到也有机会进行同步
         // 为什么要这么做呢？这里还不理解
         BuildEntitySet(ClientReplicationInfo.HandledEntities);
         BuildEntitySet(EntitiesInRange);

         // 执行到这里Entities分组完成，EntitySets中按照Entity类型存储了待同步的所有Entities
         // 注意这里的Entities并不一定就会被同步，下面还要根据LOD的计算结果再决定

         // 对EntitySets的第一次遍历，用于根据当前的Client(Viewers)更新FMassReplicationLODFragment中的LOD数据
         for (FEntitySet& Set : EntitySets)
         {
            if (Set.Entities.Num() == 0)
            {
                  continue;
            }

            // 用现成的Set设置Context
            Context.SetEntityCollection(FMassArchetypeEntityCollection(Set.Archetype, Set.Entities, FMassArchetypeEntityCollection::FoldDuplicates));

            // 通过ClientReplicationInfo将上一次处理的结果同步给Fragments
            {
                  QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_SyncToMass);
                  SyncClientData.ForEachEntityChunk(EntityManager, Context, [&ClientReplicationInfo](FMassExecutionContext& Context)
                  {
                     const TArrayView<FMassReplicationLODFragment> ViewerLODList = Context.GetMutableFragmentView<FMassReplicationLODFragment>();
                     TArrayView<FMassReplicatedAgentFragment> ReplicatedAgentList = Context.GetMutableFragmentView<FMassReplicatedAgentFragment>();

                     const int32 NumEntities = Context.GetNumEntities();
                     for (int EntityIdx = 0; EntityIdx < NumEntities; EntityIdx++)
                     {
                        FMassEntityHandle EntityHandle = Context.GetEntity(EntityIdx);
                        FMassReplicatedAgentFragment& AgentFragment = ReplicatedAgentList[EntityIdx];
                        FMassReplicationLODFragment& LODFragment = ViewerLODList[EntityIdx];

                        if (FMassReplicatedAgentData* AgentData = ClientReplicationInfo.AgentsData.Find(EntityHandle))
                        {
                              // 如果这个Entity在上一次被同步给了当前的Client直接使用上次的数据
                              LODFragment.LOD = AgentData->LOD;
                              AgentFragment.AgentData = *AgentData;
                        }
                        else
                        {
                              // 如果这个Entity在上一次没有被同步给当前的Client则清理掉Fragment中的无效数据
                              // 这里必须要清理，否则这些Fragment中的数据就是上一个客户端遗留的
                              LODFragment.LOD = EMassLOD::Off;
                              AgentFragment.AgentData.Invalidate();
                        }
                        // 总结下来，这些Fragment中只能存储一个客户端相关的数据，要么就是当前客户端的，要么就是无效的
                        // 这里有些疑惑，下面会重新计算LODFragment中的LOD，这里有必要为LOD赋值吗？可能得看看LOD的计算过程才能找到答案
                        // 看过了，LODCalculator.CalculateLOD时会用到上一次的LOD，具体计算过程此文档就不赘述了，有兴趣可自行阅读代码
                     }
                  });
            }

            // 更新Entity与Viewers的距离
            {
                  QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_LODCollection);
                  CollectViewerInfoQuery.ForEachEntityChunk(EntityManager, Context, [](FMassExecutionContext& Context)
                  {
                     const TConstArrayView<FTransformFragment> LocationList = Context.GetFragmentView<FTransformFragment>();
                     const TArrayView<FMassReplicationViewerInfoFragment> ViewersInfoList = Context.GetMutableFragmentView<FMassReplicationViewerInfoFragment>();
                     FMassReplicationSharedFragment& RepSharedFragment = Context.GetMutableSharedFragment<FMassReplicationSharedFragment>();
                     RepSharedFragment.LODCollector.CollectLODInfo(Context, LocationList, ViewersInfoList, ViewersInfoList);
                  });
            }

            // 通过上面得到的距离更新LOD
            {
                  QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_LODCaculation);
                  CalculateLODQuery.ForEachEntityChunk(EntityManager, Context, [](FMassExecutionContext& Context)
                  {
                     const TConstArrayView<FMassReplicationViewerInfoFragment> ViewersInfoList = Context.GetFragmentView<FMassReplicationViewerInfoFragment>();
                     const TArrayView<FMassReplicationLODFragment> ViewerLODList = Context.GetMutableFragmentView<FMassReplicationLODFragment>();
                     FMassReplicationSharedFragment& RepSharedFragment = Context.GetMutableSharedFragment<FMassReplicationSharedFragment>();
                     RepSharedFragment.LODCalculator.CalculateLOD(Context, ViewersInfoList, ViewerLODList, ViewersInfoList);
                  });
            }
            Context.ClearEntityCollection();
         }

         // 当然并不是所有满足LOD条件的Entities都会被同步
         // 对于每种要同步的Entities也会有数量的限制
         // 是否需要重新调整LOD将由AdjustDistancesFromCount的计算结果决定
         {
            QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_LODAdjustDistance);
            EntityManager.ForEachSharedFragment<FMassReplicationSharedFragment>([](FMassReplicationSharedFragment& RepSharedFragment)
            {
                  RepSharedFragment.bHasAdjustedDistancesFromCount = RepSharedFragment.LODCalculator.AdjustDistancesFromCount();
            });
         }

         // 再一次对EntitySets的遍历，这是最后一次了，我保证^^
         // 真正的同步就从这里面开始...
         for (FEntitySet& Set : EntitySets)
         {
            if (Set.Entities.Num() == 0)
            {
                  continue;
            }
            Context.SetEntityCollection(FMassArchetypeEntityCollection(Set.Archetype, Set.Entities, FMassArchetypeEntityCollection::FoldDuplicates));

            // 调整LOD，如果有需要的话。去代码里看一下AdjustLODDistancesQuery的设置就了然了
            {
                  QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_LODAdjustLODFromCount);
                  AdjustLODDistancesQuery.ForEachEntityChunk(EntityManager, Context, [](FMassExecutionContext& Context)
                  {
                     const TConstArrayView<FMassReplicationViewerInfoFragment> ViewersInfoList = Context.GetFragmentView<FMassReplicationViewerInfoFragment>();
                     const TArrayView<FMassReplicationLODFragment> ViewerLODList = Context.GetMutableFragmentView<FMassReplicationLODFragment>();
                     FMassReplicationSharedFragment& RepSharedFragment = Context.GetMutableSharedFragment<FMassReplicationSharedFragment>();
                     RepSharedFragment.LODCalculator.AdjustLODFromCount(Context, ViewersInfoList, ViewerLODList, ViewersInfoList);
                  });
            }

            // 这里最重要的一行代码
            // RepSharedFragment.CachedReplicator->ProcessClientReplication
            // 看看CachedReplicator的类型，终于到了UMassReplicatorBase::ProcessClientReplication
            // UMassReplicatorBase也是个工具人，她的老板原来在这里（上面是不是曾经提到过MassReplicationProcessor是个Entity网络同步的大总管，脏活累活还得交给手下去处理）
            {
                  QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_ProcessClientReplication);
                  FMassReplicationContext ReplicationContext(*World, LODSubsystem, *ReplicationSubsystem);
                  EntityManager.ForEachSharedFragment<FMassReplicationSharedFragment>([&EntityManager, &Context, &ReplicationContext, &ClientHandle](FMassReplicationSharedFragment& RepSharedFragment)
                  {
                     RepSharedFragment.CurrentClientHandle = ClientHandle;

                     RepSharedFragment.EntityQuery.ForEachEntityChunk(EntityManager, Context, [&ReplicationContext, &RepSharedFragment](FMassExecutionContext& Context)
                     {
                        RepSharedFragment.CachedReplicator->ProcessClientReplication(Context, ReplicationContext);
                     });
                  });
            }

            // 最后将同步后的结果保存到ClientReplicationInfo
            // 这里是SyncFromMass，还记得上面有个SyncToMass吗，很显然这两是一对儿
            {
                  QUICK_SCOPE_CYCLE_COUNTER(UMassReplicationProcessor_SyncFromMass);
                  SyncClientData.ForEachEntityChunk(EntityManager, Context, [&ClientReplicationInfo](FMassExecutionContext& Context)
                  {
                     TArrayView<FMassReplicatedAgentFragment> ReplicatedAgentList = Context.GetMutableFragmentView<FMassReplicatedAgentFragment>();

                     const int32 NumEntities = Context.GetNumEntities();
                     for (int EntityIdx = 0; EntityIdx < NumEntities; EntityIdx++)
                     {
                        FMassEntityHandle EntityHandle = Context.GetEntity(EntityIdx);
                        FMassReplicatedAgentFragment& AgentFragment = ReplicatedAgentList[EntityIdx];
                        ClientReplicationInfo.AgentsData.Add(EntityHandle, AgentFragment.AgentData);
                     }
                  });
            }

            // Optional debug display
            if (UE::Mass::Replication::DebugClientReplicationLOD == ClientHandle.GetIndex())
            {
                  EntityManager.ForEachSharedFragment<FMassReplicationSharedFragment>([World, &EntityManager, &Context](FMassReplicationSharedFragment& RepSharedFragment)
                  {
                     RepSharedFragment.EntityQuery.ForEachEntityChunk(EntityManager, Context, [World, &RepSharedFragment](FMassExecutionContext& Context)
                     {
                        const TConstArrayView<FTransformFragment> TransformList = Context.GetFragmentView<FTransformFragment>();
                        const TConstArrayView<FMassReplicationLODFragment> ViewerLODList = Context.GetFragmentView<FMassReplicationLODFragment>();
                        RepSharedFragment.LODCalculator.DebugDisplayLOD(Context, ViewerLODList, TransformList, World);
                     });
                  });
            }

            Context.ClearEntityCollection();
         }
      }

      // 靠，读了下面紧邻的代码终于知道了，这个是用来向客户端同步Entity删除操作的 ^^你品，你细品
      // 服务端向客户端的Entity同步有三个事件（增，删，改），如果只有当前的EntitiesInRange，我们只能知道增和改，并不能知道要删掉哪些
      // 知道上次同步了哪些，在这一次如果她不再满足LOD的条件就该被删掉了
      ClientReplicationInfo.HandledEntities = MoveTemp(EntitiesInRange);

      // 在ClientReplicationInfo中清理掉已经LOD::Off的AgentData
      for (FMassReplicationAgentDataMap::TIterator It = ClientReplicationInfo.AgentsData.CreateIterator(); It; ++It)
      {
         FMassReplicatedAgentData& AgentData = It.Value();
         if (AgentData.LOD == EMassLOD::Off)
         {
            checkf(!AgentData.Handle.IsValid(), TEXT(&#34;This replicated agent should have been removed from this client and was not&#34;));
            It.RemoveCurrent();
         }
      }
}
#endif //UE_REPLICATION_COMPILE_SERVER_CODE
}
完结撒花~
未经许可禁止转载

		自动登录	找回密码
密码			立即注册