Microsoft Orleans 之 开发人员指南

Grain接口

//anexampleofaGrainInterface
publicinterfaceIPlayerGrain:IGrainWithGuidKey
{
Task<IGameGrain>GetCurrentGame();
TaskJoinGame(IGameGraingame);
TaskLeaveGame(IGameGraingame);
}

//anexampleofaGrainclassimplementingaGrainInterface
publicclassPlayerGrain:Grain,IPlayerGrain
{
privateIGameGraincurrentGame;

//Gametheplayeriscurrentlyin.Maybenull.
publicTask<IGameGrain>GetCurrentGame()
{
returnTask.FromResult(currentGame);
}

//Gamegraincallsthismethodtonotifythattheplayerhasjoinedthegame.
publicTaskJoinGame(IGameGraingame)
{
currentGame=game;
Console.WriteLine("Player{0}joinedgame{1}",this.GetPrimaryKey(),game.GetPrimaryKey());
returnTaskDone.Done;
}

//Gamegraincallsthismethodtonotifythattheplayerhasleftthegame.
publicTaskLeaveGame(IGameGraingame)
{
currentGame=null;
Console.WriteLine("Player{0}leftgame{1}",this.GetPrimaryKey(),game.GetPrimaryKey());
returnTaskDone.Done;
}
}

//constructthegrainreferenceofaspecificplayer
IPlayerGrainplayer=GrainClient.GrainFactory.GetGrain<IPlayerGrain>(playerId);

//constructthegrainreferenceofaspecificplayer
IPlayerGrainplayer=GrainFactory.GetGrain<IPlayerGrain>(playerId);

//Invokingagrainmethodasynchronously
TaskjoinGameTask=player.JoinGame(this);
//The`await`keywordeffectivelyturnstheremainderofthemethodintoaclosurethatwillasynchronouslyexecuteuponcompletionoftheTaskbeingawaitedwithoutblockingtheexecutingthread.
awaitjoinGameTask;
//ThenextlineswillbeturnedintoaclosurebytheC#compiler.
players.Add(playerId);

List<Task>tasks=newList<Task>();
ChirperMessagechirp=CreateNewChirpMessage(text);
foreach(IChirperSubscribersubscriberinFollowers.Values)
{
tasks.Add(subscriber.NewChirpAsync(chirp));
}
TaskjoinedTask=Task.WhenAll(tasks);
awaitjoinedTask;

TaskDone.DoneUtilityProperty

客户端开发

GrainClient.Initialize();

//HardcodedplayerID
GuidplayerId=newGuid("{2349992C-860A-4EDA-9590-000000000006}");
IPlayerGrainplayer=GrainClient.GrainFactory.GetGrain<IPlayerGrain>(playerId);

IGameGraingame=player.CurrentGame.Result;
varwatcher=newGameObserver();
varobserver=GrainClient.GrainFactory.CreateObjectReference<IGameObserver>(watcher);
awaitgame.SubscribeForGameUpdates();

this.AsReference<IChirperViewer>

namespacePlayerWatcher
{
classProgram
{
///<summary>
///Simulatesacompanionapplicationthatconnectstothegame
///thataparticularplayeriscurrentlypartof,andsubscribes
///toreceivelivenotificationsaboutitsprogress.
///</summary>
staticvoidMain(string[]args)
{
try
{
GrainClient.Initialize();

//HardcodedplayerID
GuidplayerId=newGuid("{2349992C-860A-4EDA-9590-000000000006}");
IPlayerGrainplayer=GrainClient.GrainFactory.GetGrain<IPlayerGrain>(playerId);
IGameGraingame=null;

while(game==null)
{
Console.WriteLine("Gettingcurrentgameforplayer{0}...",playerId);

try
{
game=player.CurrentGame.Result;
if(game==null)//Waituntiltheplayerjoinsagame
Thread.Sleep(5000);
}
catch(Exceptionexc)
{
Console.WriteLine("Exception:",exc.GetBaseException());
}
}

Console.WriteLine("Subscribingtoupdatesforgame{0}...",game.GetPrimaryKey());

//Subscribeforupdates
varwatcher=newGameObserver();
game.SubscribeForGameUpdates(GrainClient.GrainFactory.CreateObjectReference<IGameObserver>(watcher)).Wait();

//.Waitwillblockmainthreadsothattheprocessdoesn'texit.
//Updatesarriveonthreadpoolthreads.
Console.WriteLine("Subscribedsuccessfully.Press<Enter>tostop.");
Console.ReadLine();
}
catch(Exceptionexc)
{
Console.WriteLine("UnexpectedError:{0}",exc.GetBaseException());
}
}

///<summary>
///Observerclassthatimplementstheobserverinterface.
///Needtopassagrainreferencetoaninstanceofthisclasstosubscribeforupdates.
///</summary>
privateclassGameObserver:IGameObserver
{
//Receiveupdates
publicvoidUpdateGameScore(stringscore)
{
Console.WriteLine("Newgamescore:{0}",score);
}
}
}
}

运行应用程序
配置连接到Orleans
允许应用程序与外界交互的orleansgrains，框架包括客户端库。此客户端程序库可能由桌面或移动应用程序使用，或由呈现交互式网页或公开Web服务接口的前端服务器使用。
客户端库提供了一个API编写异步客户端于orleans交互。一旦客户端库连接到orleans的网关，客户端可以发送邮件到grains，接收响应，通过观察者得到grain异步通知。连接网关
建立一个连接，客户端通过调用GrainClient.Initialize()，他将连接到silo在配置文件(ClientConfiguration.xml)中指定的ip和端口，这个文件必须和
Orleans.dll放在同一个目录下，作为一种替代，你也可以通过编程的方式加载一个配置文件然后通过[code]GrainClient.Initialize()

<ClientConfigurationxmlns="urn:orleans">
<GatewayAddress="<IPaddressorhostnameofsilo>"Port="30000"/>
</ClientConfiguration>

<?xmlversion="1.0"encoding="utf-8"?>
<OrleansConfigurationxmlns="urn:orleans">
<Defaults>
<NetworkingAddress=""Port="11111"/>
<ProxyingGatewayAddress=""Port="30000"/>
</Defaults>
</OrleansConfiguration>

grain持久化
grain持久化目标
1.允许不同的grain类型使用不同的存储类型providers（一个用Azure表，一个用SQLAzure）相同类型的存储提供程序，但具有不同的配置（两个使用Azure表，但一个使用存储帐户）。
2.允许存储提供程序实例配置可交换（例如，开发-测试-产品）只是配置文件的变化。
3.允许额外的存储供应商被写入后提供了一个框架，由orleans团队或其他人。
4.提供一个最小的生产级存储提供程序。
5.存储提供程序完全控制了在持久化支持存储中的数据状态数据的存储方式。
推论：Orleans没有提供一个全面的ORM的存储解决方案,但允在必要的时候许自定义存储提供程序支持指定一个ORM的要求。
grain持久化API
Grain类型可以用两种方式中的一种声明：
1.拓展Grain如果他们没有任何持久的状态，或如果他们将处理所有自己的持久状态，或者
2.Extend[code]Grain<T>如果他们有一些持久态，他们希望Orleans良运行时处理。

Grain<T>，因为条件限制。

[StorageProvider(ProviderName="store1")]
publicclassMyGrain<MyGrainState>...
{
...
}

<OrleansConfigurationxmlns="urn:orleans">
<Globals>
<StorageProviders>
<ProviderType="Orleans.Storage.MemoryStorage"Name="DevStore"/>
<ProviderType="Orleans.Storage.AzureTableStorage"Name="store1"
DataConnectionString="DefaultEndpointsProtocol=https;AccountName=data1;AccountKey=SOMETHING1"/>
<ProviderType="Orleans.Storage.AzureBlobStorage"Name="store2"
DataConnectionString="DefaultEndpointsProtocol=https;AccountName=data2;AccountKey=SOMETHING2"/>
</StorageProviders>

配置存储providers

AzureTableStorage

<ProviderType="Orleans.Storage.AzureTableStorage"Name="TableStore"
DataConnectionString="UseDevelopmentStorage=true"/>

DataConnectionString="..."

TableName="OrleansGrainState"

DeleteStateOnClear="false"

UseJsonFormat="false"

false

UseFullAssemblyNames="false"

UseJsonFormat="true"

false

IndentJSON="false"

UseJsonFormat="true"

false

<ProviderType="Orleans.Storage.AzureTableStorage"Name="BlobStore"
DataConnectionString="UseDevelopmentStorage=true"/>

Provider

DataConnectionString="..."

ContainerName="grainstate"

grainstate

UseFullAssemblyNames="false"

UseJsonFormat="true"

false

IndentJSON="false"

false

<ProviderType="Orleans.Storage.MemoryStorage"Name="MemoryStorage"/>

NumStorageGrains="10"

10

<ProviderType="Orleans.Storage.ShardedStorageProvider"Name="ShardedStorage">
<Provider/>
<Provider/>
<Provider/>
</Provider>

[StorageProvider]

GrainState

GrainState

Etag

string

null

Etag

任何写操作期间Orleans运行时将以grain状态数据对象的深层副本供自己使用。在重写的情况下，运行时可以使用优化规则和启发式，以避免在某些情况下，在某些情况下执行一些或所有的深拷贝，提供预期的逻辑隔离语义被保留。

Grain状态的读写操作的示例代码

为了加入Orleans的grain状态持久性机制，必须拓展Grain类。
上述定义中的T将被此grain的特定应用程序特定的grain状态类所取代；见下面的示例。
grain类也应该有一个[storageprovider]特性告诉运行时存储provider（实例）使用这种类型的grain。

publicinterfaceMyGrainState:GrainState
{
publicintField1{get;set;}
publicstringField2{get;set;}
}[StorageProvider(ProviderName="store1")]
publicclassMyPersistenceGrain:Grain<MyGrainState>,IMyPersistenceGrain
{
...
}

publicTaskDoWrite(intval)
{
State.Field1=val;
returnbase.WriteStateAsync();
}

base.ReadStateAsync()

publicasyncTask<int>DoRead()
{
awaitbase.ReadStateAsync();
returnState.Field1;
}

ReadStateAsync()

Task调用失败

WriteStateAsync()

Task

WriteStateAsync()

Task

Task

Task

WriteStateAsync()

Task

IStorageProvider。

publicinterfaceIStorageProvider
{
LoggerLog{get;}
TaskInit();
TaskClose();TaskReadStateAsync(stringgrainType,GrainIdgrainId,GrainStategrainState);
TaskWriteStateAsync(stringgrainType,GrainIdgrainId,GrainStategrainState);
}

publicclassInconsistentStateException:AggregateException
{
///<summary>TheEtagvaluecurrentlyheldinpersistentstorage.</summary>
publicstringStoredEtag{get;privateset;}
///<summary>TheEtagvaluecurrentlyheldinmemory,andattemptingtobeupdated.</summary>
publicstringCurrentEtag{get;privateset;}publicInconsistentStateException(
stringerrorMsg,
stringstoredEtag,
stringcurrentEtag,
ExceptionstorageException
):base(errorMsg,storageException)
{
this.StoredEtag=storedEtag;
this.CurrentEtag=currentEtag;
}publicInconsistentStateException(stringstoredEtag,stringcurrentEtag,ExceptionstorageException)
:this(storageException.Message,storedEtag,currentEtag,storageException)
{}
}

<OrleansConfigurationxmlns="urn:orleans">
<Globals>
<BootstrapProviders>
<ProviderType="My.App.BootstrapClass1"Name="bootstrap1"/>
<ProviderType="My.App.BootstrapClass2"Name="bootstrap2"/>
</BootstrapProviders>
</Globals>
</OrleansConfiguration>

publicvoidRegisterBootstrapProvider(stringproviderTypeFullName,stringproviderName,IDictionary<string,string>properties=null)publicvoidRegisterBootstrapProvider<T>(stringproviderName,IDictionary<string,string>properties=null)whereT:IBootstrapProvider

Orleans.Runtime.Configuration.GlobalConfiguration

TaskInit(
stringname,
IProviderRuntimeproviderRuntime,
IProviderConfigurationconfig)

protectedIDisposableRegisterTimer(Func<object,Task>asyncCallback,objectstate,TimeSpandueTime,TimeSpanperiod)

state当执行

提醒，是持久的，依赖于存储到功能。您必须指定在提醒子系统功能之前使用的存储支持。提醒功能是通过在服务器端配置的systemstore元控制。它与AzureTable或SQLServer存储一起协作。

<SystemStoreSystemStoreType="AzureTable"/>OR
<SystemStoreSystemStoreType="SqlServer"/>

<ReminderServiceReminderServiceType="ReminderTableGrain"/>

TaskIRemindable.ReceiveReminder(stringreminderName,TickStatusstatus)
{
Console.WriteLine("Thanksforremindingme--Ialmostforgot!");
returnTaskDone.Done;
}

protectedTask<IOrleansReminder>RegisterOrUpdateReminder(stringreminderName,TimeSpandueTime,TimeSpanperiod)

protectedTaskUnregisterReminder(IOrleansReminderreminder)

protectedTask<IOrleansReminder>GetReminder(stringreminderName)

Grain.OnActivateAsync

2.这几种情况下使用提醒功能

组合提醒和定时器
Youmightconsiderusingacombinationofremindersandtimerstoaccomplishyourgoal.
Forexample,ifyouneedatimerwithasmallresolutionthatneedstosurviveacrossactivations,
youcanuseareminderthatrunseveryfiveminutes,whosepurposeistowakeup
agrainthatrestartsalocaltimerthatmayhavebeenlostduetoadeactivation.

应用程序流访问APIs类似于众所周知的ReactiveExtensions(Rx)in.NET，通过使用[code]Orleans.Streams.IAsyncStream<T>

Orleans.Streams.IAsyncObserver<T>

Orleans.Streams.IAsyncObservable<T>

publicasyncTaskOnHttpCall(DeviceEventdeviceEvent)
{
//Postdatadirectlyintodevice'sstream.
IStreamProviderstreamProvider=GrainClient.GetStreamProvider("myStreamProvider");
IAsyncStream<DeviceEventData>deviceStream=streamProvider.GetStream<DeviceEventData>(deviceEvent.DeviceId);
awaitchatStream.OnNextAsync(deviceEvent.Data);
}

publicclassChatUser:Grain
{
publicasyncTaskJoinChat(stringchatGroupName)
{
IStreamProviderstreamProvider=base.GetStreamProvider("myStreamProvider");
IAsyncStream<string>chatStream=streamProvider.GetStream<string>(chatGroupName);
awaitchatStream.SubscribeAsync((stringchatEvent)=>Console.Out.Write(chatEvent));
}
}

Task

StreamSequenceToken

StreamSequenceToken

更多的例子：here.
更多的资料：