《通过C#学Proto.Actor模型》之Persistence

不管是那种持久化方式，首先要构造一个持久化的提供者，这个提者是内存也好，数据库也罢，本例中用Sqlite作为持久化的载体；在Actor中，实现持久化，首先要创建一个Persistence对象，用来将快照或事件保存起来，最重要的一点是，我们用事件溯源或快照，是帮我们保留住Actor在某刻的状，保留下来，以便我们再次启动时能延续这个状态，所以Persistence有一个很关键的作用就是能从持久化的载体中把原来的状态回复过来，这里，EventSource的把原来的状态步骤走再走一次，到达当前流程的点，但快照不然，直接取的是最后时刻的状态；带快照的事件溯源则是两者的结合。

码友看码：

usingMicrosoft.Data.Sqlite;
usingProto;
usingProto.Persistence;
usingProto.Persistence.SnapshotStrategies;
usingProto.Persistence.Sqlite;
usingSystem;
usingSystem.Threading.Tasks;

namespaceP008_Persistence
{
classProgram
{
staticvoidMain(string[]args)
{
//用sqlite持久化后
varactorid="myactorid";
vardbfile=@"C:\MyFile\Source\Repos\ProtoActorSample\ProtoActorSample\P008_Persistence\data.sqlite";
varsqliteProvider=newSqliteProvider(newSqliteConnectionStringBuilder(){DataSource=dbfile});
while(true)
{
Console.WriteLine("1、事件溯源2、快照3、带快照的事件溯源4、退出");
switch(Console.ReadLine())
{
case"1":
CallEventSource(actorid,sqliteProvider);
break;
case"2":
CallSnapShoot(actorid,sqliteProvider);
break;
case"3":
CallSnapShootEventSource(actorid,sqliteProvider);
break;
case"4":
return;
}
}
}
///<summary>
///事件溯源
///</summary>
///<paramname="actorid"></param>
///<paramname="sqliteProvider"></param>
privatestaticvoidCallEventSource(stringactorid,SqliteProvidersqliteProvider)
{
varprops=Actor.FromProducer(()=>newEventSourceDataActor(sqliteProvider,actorid));
varpid=Actor.Spawn(props);
varresult=true;
while(result)
{
Console.WriteLine("1、Tell2、删除持久化3、退出");

switch(Console.ReadLine())
{
case"1":
varrandom=newRandom();
varno=random.Next(5,15);
Console.WriteLine($"随机产生的数字：{no}");
pid.Tell(newData{Amount=no});
break;
case"2":
//完成处理后清理持久化的操作
sqliteProvider.DeleteEventsAsync(actorid,100).Wait();
break;
case"3":
result=false;
break;
}
}
}

///<summary>
///快照
///</summary>
///<paramname="actorid"></param>
///<paramname="sqliteProvider"></param>
privatestaticvoidCallSnapShoot(stringactorid,SqliteProvidersqliteProvider)
{
varprops=Actor.FromProducer(()=>newSnapShootDataActor(sqliteProvider,actorid));
varpid=Actor.Spawn(props);
varresult=true;
while(result)
{
Console.WriteLine("1、Tell2、删除持久化3、退出");

switch(Console.ReadLine())
{
case"1":
varrandom=newRandom();
varno=random.Next(5,15);
Console.WriteLine($"随机产生的数字：{no}");
pid.Tell(newData{Amount=no});
break;
case"2":
//完成处理后清理持久化的操作
sqliteProvider.DeleteEventsAsync(actorid,100).Wait();
break;
case"3":
result=false;
break;
}
}

}
///<summary>
///快照事件溯源
///</summary>
///<paramname="actorid"></param>
///<paramname="sqliteProvider"></param>
privatestaticvoidCallSnapShootEventSource(stringactorid,SqliteProvidersqliteProvider)
{
varprops=Actor.FromProducer(()=>newSnapShootEventSourceDataActor(sqliteProvider,sqliteProvider,actorid));
varpid=Actor.Spawn(props);
varresult=true;
while(result)
{
Console.WriteLine("1、Tell2、删除持久化3、退出");

switch(Console.ReadLine())
{
case"1":
varrandom=newRandom();
varno=random.Next(5,15);
Console.WriteLine($"随机产生的数字：{no}");
pid.Tell(newData{Amount=no});
break;
case"2":
//完成处理后清理持久化的操作
sqliteProvider.DeleteEventsAsync(actorid,100).Wait();
sqliteProvider.DeleteSnapshotsAsync(actorid,100).Wait();
break;
case"3":
result=false;
break;
}
}
}
}

publicclassData
{
publiclongAmount{get;set;}
}

#region事件溯源
publicclassEventSourceDataActor:IActor
{
privatelong_value=0;
privatereadonlyPersistence_persistence;

publicEventSourceDataActor(IEventStoreeventStore,stringactorId)
{
//事件溯源持久化方式
_persistence=Persistence.WithEventSourcing(eventStore,actorId,ApplyEvent);
}
privatevoidApplyEvent(Proto.Persistence.Event@event)
{
switch(@event.Data)
{
caseDatamsg:
_value=_value+msg.Amount;
Console.WriteLine($"累计：{_value}");
break;
}
}
publicasyncTaskReceiveAsync(IContextcontext)
{
switch(context.Message)
{
caseStarted_:
await_persistence.RecoverStateAsync();
break;
caseDatamsg:
await_persistence.PersistEventAsync(newData{Amount=msg.Amount});
break;
}
}
}
#endregion

#region快照
publicclassSnapShootDataActor:IActor
{
privatelong_value=0;
privatereadonlyPersistence_persistence;

publicSnapShootDataActor(ISnapshotStoresnapshotStore,stringactorId)
{
//快照持久化方式
_persistence=Persistence.WithSnapshotting(snapshotStore,actorId,ApplySnapshot);
}
privatevoidApplySnapshot(Proto.Persistence.Snapshotsnapshot)
{
switch(snapshot.State)
{
caselongvalue:
_value=value;
Console.WriteLine($"累计：{_value}");
break;
}
}
publicasyncTaskReceiveAsync(IContextcontext)
{
switch(context.Message)
{
caseStarted_:
await_persistence.RecoverStateAsync();
break;
caseDatamsg:
_value=_value+msg.Amount;
await_persistence.DeleteSnapshotsAsync(100);
await_persistence.PersistSnapshotAsync(_value);
break;
}
}
}
#endregion

#region事件溯源and快照
publicclassSnapShootEventSourceDataActor:IActor
{
privatelong_value=0;
privatereadonlyPersistence_persistence;

publicSnapShootEventSourceDataActor(IEventStoreeventStore,ISnapshotStoresnapshotStore,stringactorId)
{
//注释快照策略
//_persistence=Persistence.WithEventSourcingAndSnapshotting(eventStore,snapshotStore,actorId,ApplyEvent,ApplySnapshot,newIntervalStrategy(5),()=>{return_value;});
//无快照策略
_persistence=Persistence.WithEventSourcingAndSnapshotting(eventStore,snapshotStore,actorId,ApplyEvent,ApplySnapshot);
}
privatevoidApplyEvent(Proto.Persistence.Event@event)
{
switch(@event.Data)
{
caseDatamsg:
_value=_value+msg.Amount;
Console.WriteLine($"事件溯源累计：{_value}");
break;
}
}
privatevoidApplySnapshot(Proto.Persistence.Snapshotsnapshot)
{
switch(snapshot.State)
{
caselongvalue:
_value=value;
Console.WriteLine($"快照累计：{_value}");
break;
}
}
publicasyncTaskReceiveAsync(IContextcontext)
{
switch(context.Message)
{
caseStarted_:
await_persistence.RecoverStateAsync();
break;
caseDatamsg:
await_persistence.PersistEventAsync(newData{Amount=msg.Amount});
//无快照策略时启用
await_persistence.PersistSnapshotAsync(_value);
break;
}
}
}
#endregion
}

通过代码看到，持久化是通过在Actor中定义Persistence时，关联一个参数为，Event或Snapshot的方法，并且Actor的Receive方法在Stared到达是恢复（从持久载体中读取数据来恢复），在具体消息到达时，调用Persistence.PersistEventAsync或Persistence.PersisSnapshotAsync来持久化状态数据，这两个方法，都会把调用似递到Persistence产生是关联的那个方法，并把消息实体类通过Event.Data或Snapshot.State传递进去。

此例分别演示了事件溯源，快照，带快照事件溯源，例子很简单，就是把每次产生的随机数累加起来

1、事件溯源

三个绿色箭头，意思是进了三次“1、事件溯源”这个选项

三次蓝色箭头，意思是调用了三次Tell方法，用来获取三次随机数，蓝色椭圆是产生的三个数字，分别是7，7，6，蓝色方框是累计结果，从上往下，第一次是(0+7)7，第二次是(7+7)14，第三次是(14+6)20

红色箭头是退出事件溯源的方法，返回上一级

绿色方框是绿色箭头再次进入，自动恢复，事件溯源后的结果（即从持久化载体中把之前的所有事件重新走一次）还是之前退出时的结果，累计20，所以不管这个Actor在什么地方退出，再次运行，都会把之前的补运行回来。

也可以打开sqlite库进行查看保存的事件结果

2、快照

快照与事件溯源类似，差别在于每次再次进来，只取上次退出时的结果，同时，在数据里，只保存了最后一次的结果。

3、带快照的事件溯源

与快照类似，上面代码我们是一个事件，一个快照。

官方给出带快照的事件可以通过快照策略来保存快照

在创建持久化对象时，可以添加快照策略

_persistence=Persistence.WithEventSourcingAndSnapshotting(eventStore,snapshotStore,actorId,ApplyEvent,ApplySnapshot,newIntervalStrategy(5),()=>{return_value;});

您可以选择ISnapshotStrategy在保存事件时指定自动保存快照。提供的策略是：

同时要在Actor的Receive把保存快照注释掉，Demo中我用的是5个事件后保存一次快照，如下图结果