Orleans8.2入门测试

微软官方文档：快速入门：使用 ASP.NET Core 生成第一个 Orleans 应用 - .NET | Microsoft Learn

项目及引入的nuget库：

1、接口项目；2、接口实现项目；3、silo项目；4、客户端项目

其中Microsoft.Orleans.Streaming，为测试流功能额外添加。 

//IHello.cs代码
using System.Reflection;namespace GrainInterfaces
{/// <summary>/// Grain接口/// 从IGrainWithStringKey继承可以使用字符串作为Grain的唯一标识，便于理解/// 从IGrainObserver继承为了支持订阅方式双向通信/// </summary>public interface IHello : IGrainWithStringKey, IGrainObserver{/// <summary>/// 默认测试接口/// </summary>/// <param name="greeting"></param>/// <returns></returns>ValueTask<string> SayHello(string greeting);/// <summary>/// 传递复杂对象测试/// </summary>/// <param name="param"></param>/// <returns></returns>ValueTask<ResultA> GetResultA(ParamA param);#region 用订阅方式实现双向通信Task NotifyOtherGrain(string grainID, string message);/// <summary>/// 客户端订阅Grain上的通知消息/// 需要额外的一个IGrainObserver类型对象IChat支持/// </summary>/// <param name="observer"></param>/// <returns></returns>Task Subscribe(IChat observer);/// <summary>/// 解除订阅/// </summary>/// <param name="observer"></param>/// <returns></returns>Task Unsubscribe(IChat observer);/// <summary>/// Grains直接通信,只要有Grains的id,就可以很方便的从客户端或其他grains中拿到grains的引用/// </summary>/// <param name="sender">目标grains的id</param>/// <param name="message">要转发的消息</param>/// <returns></returns>Task ReceiveBrotherMessage(string sender, string message);#endregion}#region 订阅方式支持类定义/// <summary>/// Grain类不允许出现属性,为了方便客户端订阅,额外定义一个带Action委托的接口让Chat对象实现/// </summary>public interface IChatNotify{/// <summary>/// 收到服务端通知触发事件,客户端指定处理事件/// </summary>Action<string> ReceiveAct { get; set; }}/// <summary>/// 消息订阅支持接口/// </summary>[Alias("GrainInterfaces.IChat")]public interface IChat : IGrainObserver{/// <summary>/// 定义订阅模式消息接收函数/// 注意IChat在客户端实例化,此函数在客户端执行/// </summary>/// <param name="message"></param>/// <returns></returns>[Alias("ReceiveMessage")]Task ReceiveMessage(string message);}/// <summary>/// 消息订阅类实现(在客户端实例化)/// </summary>public class Chat : IChat, IChatNotify{public Action<string> ReceiveAct { get; set; } = null;public Task ReceiveMessage(string message){var log = $"===={message}   {Assembly.GetEntryAssembly()?.FullName}";Console.WriteLine(log);ReceiveAct?.Invoke(message);return Task.CompletedTask;}}#endregion/// <summary>/// 定义复杂函数参数类型/// 注解GenerateSerializer:告诉Orleans使用其默认的序列化反序列化方式/// </summary>[GenerateSerializer]public class ParamA{[Id(0)]public int ID { get; set; }[Id(1)]public int DataLen { get; set; }[Id(2)]public required string Name { get; set; }}/// <summary>/// 定义复杂函数返回类型/// 注解GenerateSerializer:告诉Orleans使用其默认的序列化反序列化方式/// </summary>[GenerateSerializer]public class ResultA{[Id(0)]public int ID { get; set; }[Id(1)]public required string Name { get; set; }[Id(2)]public required byte[] Data { get; set; }}/// <summary>/// 测试用全局数据定义/// </summary>public static class GlobalValueDefinition{public const string StreamProviderName = "StreamProvider";public const string GrainStorageName = "PubSubStore";public const string ImplicitStreamSubscriptionName = "RANDOMDATA";}
}

//HelloGrains.cs
using GrainInterfaces;
using Microsoft.Extensions.Logging;
using Orleans;
using Orleans.Streams;
using Orleans.Utilities;
using System.Net.NetworkInformation;namespace Grains
{/// <summary>/// Hello Grain定义/// [ImplicitStreamSubscription(GlobalValueDefinition.ImplicitStreamSubscriptionName)]/// 这个注解告诉Orleans使用隐式订阅流(简化测试)/// </summary>[ImplicitStreamSubscription(GlobalValueDefinition.ImplicitStreamSubscriptionName)]public class HelloGrains :Grain, IHello{private readonly ILogger _logger;private readonly ObserverManager<IChat> _subManager;public HelloGrains(ILogger<HelloGrains> logger){_logger = logger;_subManager = new ObserverManager<IChat>(TimeSpan.FromMinutes(5), logger);Console.WriteLine($"====={GrainContext.GrainId}");}private IAsyncStream<string> _demoStream;public override Task OnActivateAsync(CancellationToken cancellationToken){#region 流的支持CreateStreamSubscribe().GetAwaiter().GetResult();#endregionreturn base.OnActivateAsync(cancellationToken);}private async Task CreateStreamSubscribe(){var guid = this.GetPrimaryKeyString();var streamProvider = this.GetStreamProvider(GlobalValueDefinition.StreamProviderName);var streamId = StreamId.Create(GlobalValueDefinition.ImplicitStreamSubscriptionName, guid);_demoStream = streamProvider.GetStream<string>(streamId);await _demoStream.SubscribeAsync<string>(async (data, token) => { Console.WriteLine($"*****Hello {data}");await Task.CompletedTask;});}#region 订阅支持public Task Subscribe(IChat observer) {_subManager.Subscribe(observer, observer);return Task.CompletedTask;}public Task Unsubscribe(IChat observer) { _subManager.Unsubscribe(observer); return Task.CompletedTask; }//向订阅者发送消息public Task SendUpdateMessage(string message){_subManager.Notify(s=>s.ReceiveMessage(message));return Task.CompletedTask;}#endregionpublic ValueTask<string> SayHello(string greeting){var response = $"Client {GrainContext.GrainId} said:{greeting}, so HelloGrain says: Hello";_logger.LogInformation(response);SendUpdateMessage("SendUpdateMessage：" + response);return ValueTask.FromResult(response);}public ValueTask<ResultA> GetResultA(ParamA param){_logger.LogInformation($"Invoke GetResultA({param.ID}  {param.Name})");var res = new ResultA{ID = param.ID + 1,Name = param.Name,Data = new byte[param.DataLen],};return ValueTask.FromResult(res);}public Task NotifyOtherGrain(string otherGrainID, string message){var otherGrainHello = GrainFactory.GetGrain<IHello>(otherGrainID);otherGrainHello.ReceiveBrotherMessage(this.GetPrimaryKeyString(), message);return Task.CompletedTask;}public Task ReceiveBrotherMessage(string sender, string message){SendUpdateMessage($"GetBrotherMessage({sender},{message})");return Task.CompletedTask;}}
}

//Silo服务启动源码
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Orleans.Configuration;
using Orleans.Serialization;
using System.Net;
using GrainInterfaces;namespace Silo
{public partial class Form1 : Form{public Form1(){InitializeComponent();}private async void button1_Click(object sender, EventArgs e){//本地集群方式部署var primarySiloEndPoint = new IPEndPoint(IPAddress.Parse(textBox_IPAddr.Text), 11_111);var silo = new HostBuilder().UseOrleans(builder =>{builder.UseDevelopmentClustering(primarySiloEndPoint).Configure<ClusterOptions>(options =>{options.ClusterId = "cluster001";options.ServiceId = "service001";}).ConfigureEndpoints(siloPort: 11_111, gatewayPort: 30_001).ConfigureLogging(logging => logging.AddConsole()).AddMemoryStreams(GlobalValueDefinition.StreamProviderName) //unget引入 Microsoft.Orleans.Streaming.AddMemoryGrainStorage(GlobalValueDefinition.GrainStorageName);}).Build();await silo.RunAsync();}//#region 本地测试模式IHostBuilder builder = Host.CreateDefaultBuilder(args).UseOrleans(silo =>{silo.UseLocalhostClustering(11111, 30001, serviceId: "service001", clusterId: "cluster001").ConfigureLogging(logging => logging.AddConsole());}).UseConsoleLifetime();using IHost host = builder.Build();await host.RunAsync();//#endregion//#region 专用服务器集群部署//var primarySiloEndPoint = new IPEndPoint(IPAddress.Parse("192.168.0.109"), 11_111);//var silo = new HostBuilder()//    .UseOrleans(builder => //    {//        builder.UseDevelopmentClustering(primarySiloEndPoint)//        .Configure<ClusterOptions>(options =>//        {//            options.ClusterId = "cluster001";//            options.ServiceId = "service001";//        })//        .ConfigureEndpoints(siloPort: 11_111, gatewayPort: 30_001)//        .ConfigureLogging(logging => logging.AddConsole());//    }).Build();//await silo.RunAsync();//#endregion}
}

//客户端代码
using GrainInterfaces;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Orleans.Configuration;
using Orleans.Serialization.WireProtocol;
using Orleans.Streams;
using System.Net;
using System.Security.AccessControl;namespace WinFormsApp1
{public partial class Form1 : Form{public Form1(){InitializeComponent();}private IClusterClient client = null;private IHost host = null;private IHello friend1;private IHello friend2;private IChat chat;private async void button1_Click(object sender, EventArgs e){#region 开发者模式连接//IHostBuilder builder = Host.CreateDefaultBuilder()//    .UseOrleansClient(client =>//    {//        client.UseLocalhostClustering(30001, "service001", "cluster001");//    })//    .ConfigureLogging(logging => logging.AddConsole())//    .UseConsoleLifetime();//host = builder.Build();//await host.StartAsync();//client = host.Services.GetRequiredService<IClusterClient>();#endregionvar PRIMARY_SILO_IP_ADDRESS = IPAddress.Parse(textBox_IPAddr.Text);//var PRIMARY_SILO_IP_ADDRESS_1 = IPAddress.Parse("192.168.100.101");#region 专用服务器群集模式连接var gateways = new IPEndPoint[] {new IPEndPoint(PRIMARY_SILO_IP_ADDRESS, 30_001)};host = Host.CreateDefaultBuilder().UseOrleansClient(clientBuilder =>clientBuilder.UseStaticClustering(gateways).Configure<ClusterOptions>(options =>{options.ClusterId = "cluster001";options.ServiceId = "service001";})#region Stream测试.AddMemoryStreams(GlobalValueDefinition.StreamProviderName)#endregion).Build();await host.StartAsync();client = host.Services.GetRequiredService<IClusterClient>();#endregion#region 订阅friend1 = client.GetGrain<IHello>("user1");friend2 = client.GetGrain<IHello>("user2");Chat c = new Chat();(c as IChatNotify).ReceiveAct += ShowLog;chat = client.CreateObjectReference<IChat>(c);await friend1.Subscribe(chat);#endregion#region StreamTestawait CreateStreamSubscribe();#endregion}private IAsyncStream<string> _demoStream;private async Task CreateStreamSubscribe(){var clientStreamName = "clientStream001";var streamProvider = client.GetStreamProvider(GlobalValueDefinition.StreamProviderName);var streamId = StreamId.Create(GlobalValueDefinition.ImplicitStreamSubscriptionName, clientStreamName);_demoStream = streamProvider.GetStream<string>(streamId);await _demoStream.SubscribeAsync<string>(async (data, token) =>{ShowLog($"*****Hello {data}");await Task.CompletedTask;});}private async void button2_Click(object sender, EventArgs e){await host.StopAsync();}private async void button3_Click(object sender, EventArgs e){string response = await friend1.SayHello("Hi friend!");ShowLog($"{friend1.GetPrimaryKeyString()}  {response}");}private void ShowLog(string log){Invoke(new Action(() =>{textBox1.AppendText($"{log}\r\n");}));}private async void button4_Click(object sender, EventArgs e){string response = await friend2.SayHello("Hi friend!");ShowLog($"{friend2.GetPrimaryKeyString()}  {response}");}private async void button5_Click(object sender, EventArgs e){IHello friend = client.GetGrain<IHello>("user1");var dateLen = Convert.ToInt32(textBox_DataLen.Text);var result = await friend.GetResultA(new ParamA { ID = 1, Name = "henreash", DataLen = dateLen });ShowLog($"GetResultA({result.ID} {result.Name}  {result.Data.Length})");}private void button6_Click(object sender, EventArgs e){friend2.NotifyOtherGrain(friend1.GetPrimaryKeyString(), "让user2转告user1");}private async void button7_Click(object sender, EventArgs e){await _demoStream.OnNextAsync("Send test stream message!");}}
}

结论:Orleans部署不需要额外的平台,编译后的进程直接启动;通过配置即可做集群扩展.grain和grain间,grain和客户端间很方便实现双向tcp通信.关闭silo服务,客户端访问失败,启动soli服务后客户端可继续访问.

服务器增加Dashboard

silo引入nuget库OleansDashboard(8.2.0),配置增加如下内容

可方便查看silo服务器的性能数据