\u5728\u4e0a\u4e00\u7bc7\u6587\u7ae0 \u4e2d，\u6211\u6f14\u793a\u4e86 Proto.Actor（\u9002\u7528\u4e8e .NET \u7684 Actor \u6a21\u578b\u6846\u67b6）\u7684\u57fa\u7840\u7528\u6cd5\u3002\u672c\u6587\u5c06\u6784\u5efa\u4e00\u4e2a\u66f4\u590d\u6742\u7684\u793a\u4f8b：\u4f7f\u7528\u4e09\u4e2a Actor \u5b9e\u73b0 TCP \u670d\u52a1\u5668，\u5206\u522b\u5904\u7406\u8fde\u63a5\u3001\u5b57\u8282\u63a5\u6536\u548c\u6570\u636e\u5904\u7406\u3002<\/p>\n
1. WaitForTcpConnectionActor<\/p>\n
2.\u00a0ReceiveBytesActor<\/p>\n
3.\u00a0ProcessActor<\/p>\n
\u914d\u7f6e Actor \u7cfb\u7edf\u4ee5\u751f\u6210 WaitForTcpConnectionActor，\u5e76\u5728\u6309\u4e0b Ctrl+C \u65f6\u505c\u6b62\u5b83：<\/p>\n
using Proto;
\nusing TcpServer;<\/p>\n
var system = new ActorSystem();<\/p>\n
var cancellationTokenSource = new CancellationTokenSource();
\nConsole.CancelKeyPress += (_, _) =>
\n{
\n cancellationTokenSource.Cancel();
\n};<\/p>\n
system.Root.Spawn(Props.FromProducer(() => new WaitForTcpConnectionActor(9091)));<\/p>\n
while (!cancellationTokenSource.IsCancellationRequested)
\n{
\n await Task.Delay(1_000);
\n}<\/p>\n
await system.ShutdownAsync(); <\/p>\n
Actor \u6a21\u578b\u901a\u8fc7 Actor \u4e4b\u95f4\u4f20\u9012\u6d88\u606f\u5b9e\u73b0\u901a\u4fe1\u3002\u6211\u4eec\u5c06\u4f7f\u7528\u5185\u7f6e\u6d88\u606f（\u5982 Started \u548c Terminated）\u4ee5\u53ca\u81ea\u5b9a\u4e49\u6d88\u606f WaitForNextConnection\u3002<\/p>\n
\u5982\u4e0a\u4e00\u7bc7\u6587\u7ae0 \u6240\u793a，Actor \u7684\u5b9a\u4e49\u65b9\u5f0f\u5982\u4e0b：<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n public async Task ReceiveAsync(IContext context)
\n {
\n }
\n} <\/p>\n
\u7b2c\u4e00\u6b65\u662f\u542f\u52a8 TCP \u670d\u52a1\u5668\u3002\u4e3a\u6b64，\u6211\u4eec\u4f7f\u7528\u5185\u7f6e\u7684 Started \u6d88\u606f：<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n private TcpListener? _listener;<\/p>\n
public async Task ReceiveAsync(IContext context)
\n {
\n if (context.Message is Started)
\n {
\n Console.WriteLine("\u6b63\u5728 9091 \u7aef\u53e3\u76d1\u542c");
\n _listener = TcpListener.Create(port);
\n _listener.Start();
\n }
\n }
\n} <\/p>\n
\u968f\u540e，\u7b49\u5f85\u5ba2\u6237\u7aef\u8fde\u63a5\u3002\u4e3a\u6b64，\u6211\u4eec\u9700\u8981\u5411\u81ea\u8eab\u53d1\u9001\u4e00\u6761 WaitForNextConnection \u6d88\u606f：<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n private TcpListener? _listener;
\n public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is Started)
\n {
\n _listener = TcpListener.Create(port);
\n _listener.Start();
\n context.Send(context.Self, new WaitForNextConnection());
\n }
\n }
\n} <\/p>\n
\u73b0\u5728\u6211\u4eec\u5df2\u5f00\u59cb\u76d1\u542c\u8fde\u63a5，\u4e0b\u4e00\u6b65\u662f\u63a5\u6536\u5b83\u4eec，\u5e76\u4e3a\u6bcf\u4e2a\u8fde\u63a5\u751f\u6210\u65b0\u7684 ReceiveBytesActor \u5b9e\u4f8b：<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n …
\n public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is Started)
\n {
\n …
\n }
\n else if(context.Message is WaitForNextConnection)
\n {
\n var socket = await _listener!.AcceptSocketAsync(cancellationToken);
\n var actor = context.Spawn(Props.FromProducer(() => new ReceiveBytesActor()))
\n .WithChildSupervisorStrategy(new OneForOneStrategy(
\n (_, exception) =>
\n {
\n Console.WriteLine("Error: {0}", exception);
\n return SupervisorDirective.Restart;
\n },
\n 3,
\n TimeSpan.FromSeconds(1)));;
\n context.Send(actor, new SocketAccepted(socket));
\n context.Send(context.Self, new WaitForNextConnection());
\n }
\n }
\n} <\/p>\n
\u6211\u4eec\u901a\u8fc7 OneForOneStrategy \u914d\u7f6e\u5bf9 ReceiveBytesActor \u5b9e\u4f8b\u7684\u76d1\u7763\u7b56\u7565：<\/p>\n
\u5f53\u5904\u7406\u5b8c\u6210\u540e，\u5b50 Actor \u4f1a\u5411\u7236 Actor \u53d1\u9001 ProcessCompleted \u6d88\u606f\u3002\u8fd9\u63d0\u793a\u7236 Actor \u663e\u5f0f\u505c\u6b62\u5b50 Actor ，\u786e\u4fdd\u8d44\u6e90\u6e05\u7406\u5e76\u907f\u514d\u5185\u5b58\u6cc4\u6f0f\u3002<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n …
\n public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is Started)
\n {
\n …
\n }
\n else if(context.Message is { Message: Terminated, Sender: not null }))
\n {
\n _listener?.Dispose();
\n }
\n else if(context.Message is ProcessCompleted)
\n {
\n await context.StopAsync(Sender);
\n }
\n else if(context.Message is WaitForNextConnection)
\n {
\n …
\n }
\n }
\n} <\/p>\n
\u5f53\u8fde\u63a5\u5904\u7406\u5b8c\u6210\u540e：<\/p>\n
using System.Net.Sockets;
\nusing Proto;<\/p>\n
namespace TcpServer;<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n private static readonly Props Props = Props.FromProducer(() => new ReceiveBytesActor())
\n .WithChildSupervisorStrategy(new OneForOneStrategy(
\n (_, exception) =>
\n {
\n Console.WriteLine("Error: {0}", exception);
\n return SupervisorDirective.Restart;
\n },
\n 3,
\n TimeSpan.FromSeconds(1)));<\/p>\n
private TcpListener? _listener;<\/p>\n
public async Task ReceiveAsync(IContext context)
\n {
\n if (context is { Message: Terminated, Sender: not null })
\n {
\n Close();
\n }
\n else if (context.Message is Started)
\n {
\n Open();
\n context.Send(context.Self, new WaitForNextConnection());
\n }
\n else if (context.Message is ProcessCompleted)
\n {
\n Console.WriteLine("stopping actor: {0}", context.Sender);
\n await context.StopAsync(context.Sender!);
\n }
\n else if(context.Message is WaitForNextConnection)
\n {
\n var socket = await AcceptTcpClientAsync(context.CancellationToken);
\n Console.WriteLine("Accepted connection from {0}", socket.RemoteEndPoint);
\n var actor = context.Spawn(Props);
\n context.Send(actor, new SocketAccepted(socket));
\n context.Send(context.Self, new WaitForNextConnection());
\n }
\n }<\/p>\n
private void Open()
\n {
\n Close();
\n Console.WriteLine("Listening on port 9091");
\n _listener = TcpListener.Create(port);
\n _listener.Start();
\n }<\/p>\n
private async Task<Socket> AcceptTcpClientAsync(CancellationToken cancellationToken)
\n {
\n Console.WriteLine("Waiting for connection…");
\n return await _listener!.AcceptSocketAsync(cancellationToken);
\n }<\/p>\n
private void Close()
\n {
\n Console.WriteLine("Closing listener");
\n _listener?.Dispose();
\n }
\n} <\/p>\n
\u5f53 Actor \u7cfb\u7edf\u5173\u95ed\u65f6，\u5fc5\u987b\u6b63\u786e\u91ca\u653e TCP \u76d1\u542c\u5668 ，\u4ee5\u907f\u514d\u8d44\u6e90\u6cc4\u6f0f：<\/p>\n
public class WaitForTcpConnectionActor(int port) : IActor
\n{
\n …
\n public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is Started)
\n {
\n …
\n }
\n else if(context.Message is { Message: Terminated, Sender: not null }))
\n {
\n _listener?.Dispose();
\n }
\n else if(context.Message is ProcessCompleted)
\n {
\n ….
\n }
\n else if(context.Message is WaitForNextConnection)
\n {
\n …
\n }
\n }
\n} <\/p>\n
\u4e0b\u4e00\u6b65\u662f\u4ece\u5957\u63a5\u5b57\u63a5\u6536\u5b57\u8282\u6d41：<\/p>\n
\u5f53\u65b0\u8fde\u63a5\u88ab\u63a5\u53d7\u65f6，Actor \u4f1a\u5b58\u50a8\u5957\u63a5\u5b57\u5b9e\u4f8b\u5e76\u8bfb\u53d6\u53ef\u7528\u5b57\u8282：<\/p>\n
public class ReceiveBytesActor : IActor
\n{
\n private Socket? _socket;
\n private byte[]? _buffer;
\n public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is SocketAccepted socket)
\n {
\n _socket = socket;
\n _buffer = new byte[_socket.Available];
\n await _socket.ReceiveAsync(_buffer);<\/p>\n
var props = Props.FromProducer(() => new ProcessActor());
\n var actor = context.SpawnNamed(props, "json-serializer");
\n context.Send(actor, new SocketReceived(_buffer!));
\n }
\n }
\n} <\/p>\n
\u5904\u7406\u5b8c\u6210\u540e，\u5f53\u524d Actor \u505c\u6b62\u5b50\u7ea7 ProcessActor \u5e76 \u901a\u77e5\u5176\u7236\u7ea7\u91ca\u653e\u8d44\u6e90 ：<\/p>\n
public class ReceiveBytesActor : IActor
\n{
\n …<\/p>\n
public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is SocketAccepted socket)
\n {
\n …
\n }
\n else if(context.Message is ProcessCompleted)
\n {
\n await context.StopAsync(Sender);
\n context.Send(context.Parent!, new ProcessCompleted());
\n }
\n }
\n} <\/p>\n
\u5f53 Actor \u7ec8\u6b62\u65f6，\u91ca\u653e\u5957\u63a5\u5b57\u8d44\u6e90 \u5e76\u505c\u6b62\u6240\u6709\u5b50\u7ea7 Actor \u4ee5\u9632\u6b62\u8d44\u6e90\u6cc4\u6f0f：<\/p>\n
public class ReceiveBytesActor : IActor
\n{
\n …<\/p>\n
public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is Terminated)
\n {
\n _buffer = null;
\n _socket?.Dispose();
\n await context.Children.StopMany(context);
\n }
\n else if(context.Message is SocketAccepted socket)
\n {
\n …
\n }
\n else if(context.Message is ProcessCompleted)
\n {
\n …
\n }
\n }
\n} <\/p>\n
\u82e5 ProcessActor \u53d1\u751f\u6545\u969c\u5e76\u91cd\u542f，ReceiveBytesActor \u4f1a\u91cd\u65b0\u53d1\u9001\u5176\u7f13\u51b2\u533a\u4e2d\u7684\u6570\u636e\u4ee5\u91cd\u65b0\u5904\u7406：<\/p>\n
public class ReceiveBytesActor : IActor
\n{
\n …<\/p>\n
public async Task ReceiveAsync(IContext context)
\n {
\n if(context.Message is Terminated)
\n {
\n …
\n }
\n else if(context.Message is SocketAccepted socket)
\n {
\n …
\n }
\n else if(context.Message is ProcessCompleted)
\n {
\n …
\n }
\n else if(context.Message is ResendBufferReceived)
\n {
\n context.Send(Sender, new ResendBufferReceived(_buffer!));
\n }
\n }
\n} <\/p>\n
using System.Net.Sockets;
\nusing Proto;<\/p>\n
namespace TcpServer;<\/p>\n
public class ReceiveBytesActor : IActor
\n{
\n private Socket? _socket;
\n private byte[]? _buffer;<\/p>\n
public async Task ReceiveAsync(IContext context)
\n {
\n if (context.Message is Terminated)
\n {
\n Console.WriteLine("Terminating actor");
\n _buffer = null;
\n _socket?.Dispose();
\n await context.Children.StopMany(context);
\n }
\n else if (context.Message is ProcessCompleted)
\n {
\n Console.WriteLine("Process completed");
\n await context.StopAsync(context.Sender!);
\n context.Send(context.Parent!, new ProcessCompleted());
\n }
\n else if (context.Message is SocketAccepted socketAccepted)
\n {
\n await SocketAccepted(socketAccepted.Socket);<\/p>\n
var props = Props.FromProducer(() => new ProcessActor());
\n var actor = context.SpawnNamed(props, "json-serializer");
\n context.Send(actor, new BufferReceived(_buffer!));
\n }
\n else if (context.Message is ResendBufferReceived)
\n {
\n context.Send(context.Sender!, new BufferReceived(_buffer!));
\n }
\n }<\/p>\n
private async Task SocketAccepted(Socket socket)
\n {
\n if (_socket != null)
\n {
\n return;
\n }<\/p>\n
_socket = socket;
\n _buffer = new byte[_socket.Available];
\n await _socket.ReceiveAsync(_buffer);
\n }
\n} <\/p>\n
\u6700\u7ec8\u7684 Actor \u8d1f\u8d23\u53cd\u5e8f\u5217\u5316\u6570\u636e\u5e76\u8bb0\u5f55\u65e5\u5fd7：<\/p>\n
BufferReceived \u6d88\u606f\u5305\u542b\u4ece\u5957\u63a5\u5b57\u63a5\u6536\u7684\u539f\u59cb\u5b57\u8282\u6d41\u3002\u8be5 Actor \u5c06\u6570\u636e\u53cd\u5e8f\u5217\u5316\u4e3a Sample \u5bf9\u8c61\u5e76\u8f93\u51fa\u5230\u63a7\u5236\u53f0\u3002\u5904\u7406\u5b8c\u6210\u540e，\u901a\u8fc7 ProcessCompleted \u6d88\u606f\u901a\u77e5\u7236\u7ea7 Actor（ReceiveBytesActor）\u6e05\u7406\u8d44\u6e90：<\/p>\n
public class ProcessActor : IActor
\n{
\n public Task ReceiveAsync(IContext context)
\n {
\n if (context.Message is BufferReceived socketReceived)
\n {
\n var json = JsonSerializer.Deserialize<Sample>(socketReceived.Data)!;
\n Console.WriteLine("Received sample with id: {0} and name: {1}", json.Id, json.Name);
\n context.Send(context.Parent!, new ProcessCompleted(context.Self));
\n }
\n return Task.CompletedTask;
\n }
\n} <\/p>\n
\u5f53 Actor \u53d1\u751f\u6545\u969c\u5e76\u91cd\u542f\u65f6，Proto.Actor \u4f1a\u5411\u8be5 Actor \u53d1\u9001 Restarting \u6d88\u606f\u3002\u8fd9\u4f7f\u5f97\u91cd\u542f\u7684 Actor \u80fd\u591f\u901a\u77e5\u5176\u7236\u7ea7\u91cd\u65b0\u4f20\u8f93\u539f\u59cb\u6d88\u606f（\u6216\u72b6\u6001），\u4ee5\u4fbf\u91cd\u542f\u540e\u7684 Actor \u53ef\u4ee5\u91cd\u65b0\u5904\u7406\u8fd9\u4e9b\u6570\u636e<\/p>\n
public class ProcessActor : IActor
\n{
\n public Task ReceiveAsync(IContext context)
\n {
\n if (context.Message is Restarting)
\n {
\n context.Send(context.Parent!, new ResendBufferReceived());
\n }
\n else if (context.Message is BufferReceived socketReceived)
\n {
\n var json = JsonSerializer.Deserialize<Sample>(socketReceived.Data)!;
\n Console.WriteLine("Received sample with id: {0} and name: {1}", json.Id, json.Name);
\n context.Send(context.Parent!, new ProcessCompleted(context.Self));
\n }
\n return Task.CompletedTask;
\n }
\n} <\/p>\n
using System.Text.Json;
\nusing Proto;<\/p>\n
namespace TcpServer;<\/p>\n
public class ProcessActor : IActor
\n{
\n public Task ReceiveAsync(IContext context)
\n {
\n if (context.Message is Restarting)
\n {
\n context.Send(context.Parent!, new ResendBufferReceived());
\n }
\n else if (context.Message is BufferReceived socketReceived)
\n {
\n var json = JsonSerializer.Deserialize<Sample>(socketReceived.Data)!;
\n Console.WriteLine("Received sample with id: {0} and name: {1}", json.Id, json.Name);
\n context.Send(context.Parent!, new ProcessCompleted());
\n }
\n return Task.CompletedTask;
\n }
\n} <\/p>\n
\u6700\u540e，\u6211\u4eec\u6765\u5b9e\u73b0\u4e00\u4e2a\u7b80\u5355\u7684 TCP \u5ba2\u6237\u7aef，\u5b83\u4f1a\u5c06\u7528\u6237\u8f93\u5165\u8f6c\u6362\u4e3a JSON \u683c\u5f0f\u5e76\u53d1\u9001\u5230\u670d\u52a1\u5668：<\/p>\n
using System.Net.Sockets;
\nusing System.Text.Json;
\nusing TcpServer.Client;<\/p>\n
var id = 0;
\nwhile (true)
\n{
\n Console.Write("Type a name (q to quit\/f to non json): ");
\n var name = Console.ReadLine();
\n if (string.IsNullOrWhiteSpace(name))
\n {
\n continue;
\n }<\/p>\n
if (name == "q")
\n {
\n break;
\n }<\/p>\n
try
\n {
\n var connection = new TcpClient();
\n await connection.ConnectAsync("localhost", 9091);<\/p>\n
var stream = connection.GetStream();
\n if (name == "f")
\n {
\n await stream.WriteAsync(new[] { (byte)'f' });
\n }
\n else
\n {
\n await JsonSerializer.SerializeAsync(stream, new Sample
\n {
\n Id = id++,
\n Name = name,
\n });
\n }
\n connection.Close();
\n }
\n catch (Exception e)
\n {
\n Console.WriteLine("Error: {0}",e);
\n }
\n} <\/p>\n
Proto.Actor \u5e93\u662f\u6784\u5efa\u5bb9\u9519\u7cfb\u7edf\u7684\u5f3a\u5927\u5de5\u5177，\u5b83\u5c55\u793a\u4e86 Actor \u6a21\u578b\u5982\u4f55\u7b80\u5316\u5e76\u53d1\u5904\u7406\u3001\u8d44\u6e90\u7ba1\u7406\u548c\u9519\u8bef\u6062\u590d\u3002\u901a\u8fc7\u4f7f\u7528\u4e09\u4e2a Actor（WaitForTcpConnectionActor\u3001ReceiveBytesActor \u548c ProcessActor）\u5b9e\u73b0 TCP \u670d\u52a1\u5668，\u6211\u4eec\u63a2\u7d22\u4e86\u4ee5\u4e0b\u6838\u5fc3\u6982\u5ff5：<\/p>\n
\u6838\u5fc3\u5b9e\u8df5<\/p>\n
\u672c\u793a\u4f8b\u4e3a\u6559\u5b66\u76ee\u7684\u7b80\u5316\u4e86\u590d\u6742\u573a\u666f\u3002\u4f8b\u5982：<\/p>\n
_socket.Available \u7684\u5c40\u9650\u6027<\/p>\n
\u9519\u8bef\u5904\u7406\u7684\u6539\u8fdb<\/p>\n
\u8d44\u6e90\u6e05\u7406\u7684\u589e\u5f3a<\/p>\n
GitHub \u4ed3\u5e93<\/p>\n","protected":false},"excerpt":{"rendered":"
\u6587\u7ae0\u6d4f\u89c8\u9605\u8bfb789\u6b21\uff0c\u70b9\u8d5e24\u6b21\uff0c\u6536\u85cf8\u6b21\u3002\u672c\u6587\u4ecb\u7ecd\u4e86\u5982\u4f55\u4f7f\u7528Proto.Actor\u6846\u67b6\u6784\u5efa\u4e00\u4e2a\u590d\u6742\u7684TCP\u670d\u52a1\u5668\uff0c\u901a\u8fc7\u4e09\u4e2aActor\u5206\u522b\u5904\u7406\u8fde\u63a5\u3001\u5b57\u8282\u63a5\u6536\u548c\u6570\u636e\u5904\u7406\u3002\u6838\u5fc3Actor\u5305\u62ecWaitForTcpConnectionActor\u3001ReceiveBytesActor\u548cProcessActor\uff0c\u5206\u522b\u8d1f\u8d23\u76d1\u542c\u8fde\u63a5\u3001\u63a5\u6536\u5b57\u8282\u6d41\u548c\u53cd\u5e8f\u5217\u5316\u6570\u636e\u3002\u6587\u7ae0\u8be6\u7ec6\u8bf4\u660e\u4e86\u5982\u4f55\u914d\u7f6eActor\u7cfb\u7edf\u3001\u5b9e\u73b0TCP\u76d1\u542c\u3001\u5904\u7406\u8fde\u63a5\u3001\u63a5\u6536\u5b57\u8282\u3001\u53cd\u5e8f\u5217\u5316\u6570\u636e\u4ee5\u53ca\u8d44\u6e90\u6e05\u7406\u3002\u6b64\u5916\uff0c\u8fd8\u63a2\u8ba8\u4e86Actor\u76d1\u7763\u673a\u5236\u3001\u6d88\u606f\u5f39\u6027\u3001\u751f\u547d\u5468\u671f\u7ba1\u7406\u7b49\u6838\u5fc3\u6982\u5ff5\uff0c\u5e76\u63d0\u4f9b\u4e86\u751f\u4ea7\u73af\u5883\u4e2d\u7684\u4f18\u5316\u5efa\u8bae\u3002\u901a\u8fc7\u8fd9\u4e2a\u793a\u4f8b\uff0c<\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[895,3612,3613,55,3611,205],"topic":[],"class_list":["post-39842","post","type-post","status-publish","format-standard","hentry","category-server","tag-net","tag-actor-model","tag-actor-","tag-c","tag-proto-actor","tag-tcp-ip"],"yoast_head":"\n