Orleans解决并发之痛(四):Streams

Orleans 提供了 Stream 扩展编程模型。此模型提供了一套 API,使处理流更简单和更健壮。Stream 默认提供了两种 Provider,Simple Message Stream ProviderAzure Queue Stream Provider,不同的流类型可能使用不同的 Provider 来处理。Stream Providers 兼容现有的队列技术,比如: Event HubsServiceBusAzure QueuesApache Kafka,不需要编写额外的代码来配合这些队列技术的使用。

关于为什么Orleans会提供Stream扩展编程模型?

当今已经有一系列技术可以来构建一个流处理系统。包括持久存储流数据方面,如:Event HubsKafka;数据流计算操作方面,如: Azure Stream AnalyticsApache StormApache Spark Streaming, 而这些技术并不适合细粒度的自由格式的流数据计算, 或者支持的并不好,因为实际情况下可能需要对不同的数据流执行不同的操作,Orleans Streams目的就是解决这类问题,Stream 编程模型和发布订阅模式挺相似。

上述提到的一些技术我并没有详细学习,后面会了解并对比。

Orleans Stream 大概实现的步骤如下:

  1. 获取 StreamProvider
  2. 获取 IAsyncStream
  3. 订阅者订阅一个Stream
  4. 发布者向某个Stream发布消息

Silo 配置文件 OrleansConfiguration.xml 修改

在Globals节点中添加:

1
2
3
4
5
6
<StorageProviders>
<Provider Type="Orleans.Storage.MemoryStorage" Name="PubSubStore" />
</StorageProviders>
<StreamProviders>
<Provider Type="Orleans.Providers.Streams.SimpleMessageStream.SimpleMessageStreamProvider" Name="SMSProvider"/>
</StreamProviders>

Name 为 PubSubStore(名字任意) 的 StorageProvider 是必须的,Stream 内部需要它来跟踪所有流订阅,记录各个流的发布者和订阅者的关系,本例测试使用 MemoryStorage,实际生产环境不能使用 Memory 方式。

Name 为 SMSProvider(名字任意)的 StreamProvider 指定了消息的发布形式,Orleans 当前提供的两种 StreamProvider:Simple Message Stream ProviderAzure Queue Stream Provider 都是可靠的。

Simple Message Stream Provider:不保证可靠的交付,失败的消息不会自动重新发送,但可以根据返回的Task状态来判断是否重新发送,事件执行顺序遵循 FIFO 原则。

Azure Queue Stream Provider:事件被加入 Azure Queue, 如果传送或处理失败,事件不会从队列中删除,并且稍后会自动重新被发送,因此事件执行顺序不遵循 FIFO 原则。

获取 StreamProvider

1
var streamProvider = this.GetStreamProvider("SMSProvider");

SMSProvider 对应配置文件中 Name 为 SMSProvider 的 StreamProvider

获取 IAsyncStream

1
2
var streamId = this.GetPrimaryKey();
var stream = streamProvider.GetStream<string>(streamId, "GrainStream");

GetStream 需要两个参数,通过两个值定位唯一的 Stream:
streamId:guid 类型,stream 标识
streamNamespace:字符串,stream 的命名空间

订阅一个Stream

订阅 Stream 分为隐式和显式订阅。

隐式订阅

隐式订阅模式订阅者自动由发布者创建,订阅者是唯一的,不存在对一个 Stream 的多次订阅,也不可手动取消订阅。

Interface:

1
2
3
public interface IImplicitSubscriberGrain : IGrainWithGuidKey
{
}

Grain:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
[ImplicitStreamSubscription("GrainImplicitStream")]
public class ImplicitSubscriberGrain : Grain, IImplicitSubscriberGrain, IAsyncObserver<string>
{
protected StreamSubscriptionHandle<string> streamHandle;

public override async Task OnActivateAsync()
{
var streamId = this.GetPrimaryKey();
var streamProvider = this.GetStreamProvider("SMSProvider");
var stream = streamProvider.GetStream<string>(streamId, "GrainImplicitStream");
streamHandle = await stream.SubscribeAsync(OnNextAsync);
}

public override async Task OnDeactivateAsync()
{
if (streamHandle != null)
await streamHandle.UnsubscribeAsync();
}

public Task OnCompletedAsync()
{
return Task.CompletedTask;
}

public Task OnErrorAsync(Exception ex)
{
return Task.CompletedTask;
}

public Task OnNextAsync(string item, StreamSequenceToken token = null)
{
Console.WriteLine($"Received message:{item}");
return Task.CompletedTask;
}
}

  1. 在 Grain 上标记 ImplicitStreamSubscription 属性,变量值为命名空间;
  2. 在 Grain 的 OnActivateAsync 方法体中调用 SubscribeAsync ,Grain 创建即订阅;
  3. 实现 IAsyncObserver 接口,当发布者向 Stream 发送消息,订阅者接到消息后将执行 OnNextAsync;

显式订阅

Interface:

1
2
3
4
5
6
public interface IExplicitSubscriberGrain : IGrainWithGuidKey
{
Task<StreamSubscriptionHandle<string>> SubscribeAsync();

Task ReceivedMessageAsync(string data);
}

Grain:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class ExplicitSubscriberGrain : Grain, IExplicitSubscriberGrain
{
private IAsyncStream<string> stream;

public async override Task OnActivateAsync()
{
var streamProvider = this.GetStreamProvider("SMSProvider");
stream = streamProvider.GetStream<string>(this.GetPrimaryKey(), "GrainExplicitStream");
var subscriptionHandles = await stream.GetAllSubscriptionHandles();
if (subscriptionHandles.Count > 0)
{
subscriptionHandles.ToList().ForEach(async x =>
{
await x.ResumeAsync((payload, token) => this.ReceivedMessageAsync(payload));
});
}
}

public async Task<StreamSubscriptionHandle<string>> SubscribeAsync()
{
return await stream.SubscribeAsync((payload, token) => this.ReceivedMessageAsync(payload));
}

public Task ReceivedMessageAsync(string data)
{
Console.WriteLine($"Received message:{data}");
return Task.CompletedTask;
}
}

  1. 订阅者通过调用 SubscribeAsync 方法完成订阅,并返回 StreamSubscriptionHandle ,这个对象提供了 UnsubscribeAsync 方法,支持手动取消订阅;

  2. 本例子中支持对同一个 Stream 订阅多次,被订阅多次的结果是当向这个 Stream 发送消息的时候,ReceivedMessageAsync 会执行多次。如果不希望对同一个 Stream 订阅多次,在 SubscribeAsync 方法中可以通过GetAllSubscriptionHandles 获取当前已订阅者的数量,然后进行限制;

  3. 订阅者是一直存在的,除了被显示调用了 UnsubscribeAsync 方法。在 OnActivateAsync 中我们加入了 ResumeAsync 操作, 当 Grain 由未激活状态变为激活状态的时候,先通过 GetAllSubscriptionHandles 获取这个 Stream 中存在的订阅者,然后通过 ResumeAsync 可以把它们重新唤醒。(模拟方式:杀掉Silo,重新启动即可,不过前提条件是 PubSubStore 不能使用 MemoryStorage ,因为使用 MemoryStorage 存储一旦重启后订阅者和发布者的关系都会丢失)

发布消息

Interface:

1
2
3
4
public interface IPublisherGrain: IGrainWithGuidKey
{
Task PublishMessageAsync(string data);
}

Grain:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class PublisherGrain : Grain, IPublisherGrain
{
private IAsyncStream<string> stream;

public override Task OnActivateAsync()
{
var streamId = this.GetPrimaryKey();
var streamProvider = this.GetStreamProvider("SMSProvider");
this.stream = streamProvider.GetStream<string>(streamId, "GrainExplicitStream"); //隐式:GrainImplicitStream
return base.OnActivateAsync();
}

public async Task PublishMessageAsync(string data)
{
Console.WriteLine($"Sending data: {data}");
await this.stream.OnNextAsync(data);
}
}

通过调用 IAsyncStream 的 OnNextAsync 发布消息即可。这里可以针对返回的 Task 状态再作一些操作,如果不成功,重新发送或记录日志等。

Client 发布消息:

客户端发布消息:

1
2
3
4
5
6
7
8
while (true)
{
Console.WriteLine("Press 'exit' to exit...");
var input = Console.ReadLine();
if (input == "exit") break;
var publisherGrain = GrainClient.GrainFactory.GetGrain<IPublisherGrain>(Guid.Empty);
publisherGrain.PublishMessageAsync(input);
}

发布消息

显示订阅下,需要增加另一个客户端先完成订阅,可启动多个来创建多个订阅者:

1
2
3
4
5
var subscriberGrain = GrainClient.GrainFactory.GetGrain<IExplicitSubscriberGrain>(Guid.Empty);
var streamHandle = subscriberGrain.SubscribeAsync().Result;
Console.WriteLine("Press enter to exit...");
Console.ReadLine();
streamHandle.UnsubscribeAsync();

显示订阅下发布消息

参考链接:

如果对你有帮助就好