Akka.NET
Akka.NET is an open source, cross-platform actor model toolkit for building high-performance, scalable, and fault-tolerant distributed applications. It is a .NET version of the Akka framework on Java and aims to provide the same benefits and capabilities to .NET developers.
Akka.NET provides a concurrency model called the Actor model. An Actor is a lightweight processing unit that has its own state and behavior and communicates through messages. They are highly independent, can execute concurrently, and can communicate through message passing, enabling distributed communication within the application and across the network.
Akka.NET provides rich functionality, including the following:
- Actor model: Akka.NET provides a powerful actor model for building high-performance concurrent applications. Actors communicate with each other through message passing, enabling decoupling and concurrent processing.
- Distributed Systems: Akka.NET has built-in distributed features that make it easy to build distributed systems. It provides message routing, cluster management, fault recovery and other functions, making the development of distributed systems simple and reliable.
- Fault tolerance: Akka.NET provides a powerful fault tolerance mechanism that enables applications to automatically recover and handle failures. It adopts a supervision model where parent actors are responsible for monitoring and managing child actors to ensure system reliability and resilience.
- Concurrency and performance: Akka.NET’s actor model and asynchronous messaging mechanism enable highly concurrent and high-performance applications. It can take advantage of multi-core and distributed environments and perform efficient resource utilization.
Next, let’s use a simple example to build an additional Actor model.
1: Install Akka.net. We install these 3 packages under NuGet.
Akka.Remote is a module in the Akka.NET framework used to implement distributed scenarios. It allows different Akka systems to communicate over the network, enabling message passing between actors across hosts. Akka.Remote provides a transparent way for local actors to send and receive messages like local actors, while solving the complexity of cross-node communication.
Based on Akka.Remote, you can build distributed applications where actors can be located on different hosts and communicate through asynchronous messaging. Akka.Remote provides highly reliable network communication support, handles network failures, message delivery failures, etc., and provides a configurable retry mechanism and message delivery guarantees. It also allows applications to dynamically connect and disconnect from remote nodes at runtime.
Akka.Cluster allows multiple Akka.NET nodes to be formed into a cluster, and the nodes can be located on multiple hosts. These nodes can automatically discover and join the cluster, and communicate via Akka.Remote. Akka.Cluster uses a distributed consensus algorithm based on the gossip protocol to maintain state synchronization and consistency among nodes in the cluster. It provides fault detection and failover capabilities, allowing automatic restart and redistribution of tasks when a node fails. In addition, Akka.Cluster also provides configurable routing and load balancing mechanisms that can be flexibly adjusted according to actual needs.
using System;
using Akka.Actor;
namespace AkkaExample
{<!-- -->
//Define a simple message class
public class GreetingMessage
{<!-- -->
public string Message {<!-- --> get; private set; }
public GreetingMessage(string message)
{<!-- -->
Message = message;
}
}
//Define an Actor class
public class GreetingActor : ReceiveActor
{<!-- -->
publicGreetingActor()
{<!-- -->
// Declare the message types that the Actor can handle
Receive<GreetingMessage>(message =>
{<!-- -->
Console.WriteLine("Received message: " + message.Message);
});
}
}
class Program
{<!-- -->
static void Main(string[] args)
{<!-- -->
//Create an ActorSystem
var system = ActorSystem.Create("GreetingSystem");
//Create a GreetingActor instance
var greetingActor = system.ActorOf<GreetingActor>("GreetingActor");
//Send message to GreetingActor
greetingActor.Tell(new GreetingMessage("Hello, Akka.NET!"));
// Wait for any key to be pressed to exit
Console.Read();
// Stop ActorSystem
system.Terminate().Wait();
}
}
}
Detailed explanation of Akka.Cluster configuration
var config = ConfigurationFactory.ParseString(@"
akka {
loglevel = debug #Set the log level to debug, optional values: off, error, warning, info, debug
loggers = [""Akka.Logger.NLog.NLogLogger, Akka.Logger.NLog""] # Set the log output to NLog
actor {
provider = remote # Set the provider of the Actor system to remote
creation-timeout = 10s #Set the Actor creation timeout to 10 seconds
debug {
receive = on # Enable debug logging of received messages
autoreceive = on # Enable debugging logs for automatically receiving messages
lifecycle = on # Enable debugging logging of life cycle events
event-stream = on # Enable debug logging of event streams
unhandled = on # Enable debug logging of unhandled messages
}
}
remote {
dot-netty.tcp {
#Set the remote communication protocol to TCP, optional values: tcp, udp
hostname = localhost #Set the bound IP address
port = 8080 #Set the bound port number
}
# Customize remote communication protocol related settings (optional)
my-protocol {
# Fill in the configuration of the custom protocol here
}
}
}
");
//Create the Actor system and use the configuration
var actorSystem = ActorSystem.Create("MyActorSystem", config);
With this foundation, we first draw the server architecture diagram
DotNetty
using DotNetty.Buffers;
using DotNetty.Handlers.Timeout;
using DotNetty.Transport.Bootstrapping;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Sockets;
using System;
using System.Threading.Tasks;
namespace KuaFuServer.Net
{<!-- -->
public static class TcpServer
{<!-- -->
private static IEventLoopGroup bossGroup;
private static IEventLoopGroup workerGroup;
private static IChannel bootstrapChannel;
/// <summary>
/// Start TcpServer
/// </summary>
/// <param name="port"></param>
/// <returns></returns>
public static Task Start(int port)
{<!-- -->
return RunServerAsync(port);
}
private static async Task RunServerAsync(int port)
{<!-- -->
// Main working group
bossGroup = new MultithreadEventLoopGroup(1);
//The default is the number of CPU cores * 2
workerGroup = new MultithreadEventLoopGroup();
try
{<!-- -->
ServerBootstrap bootstrap = new ServerBootstrap();
//Set the thread group model to: master-slave thread model
bootstrap.Group(bossGroup, workerGroup);
//Set the channel type to TCP
bootstrap.Channel<TcpServerSocketChannel>();
bootstrap
//The upper limit of the elements of the semi-connection queue means that three handshakes have been completed at the operating system level and the number of links waiting for the current process to be taken away
.Option(ChannelOption.SoBacklog, 1024)//Set the queue size of TCP connection
// Used to set the buffer size when Channel receives byte stream
.Option(ChannelOption.RcvbufAllocator, new AdaptiveRecvByteBufAllocator()) Adaptive receive buffer allocator
.Option(ChannelOption.Allocator, PooledByteBufferAllocator.Default) // Use heap memory allocator
.ChildOption(ChannelOption.Allocator, PooledByteBufferAllocator.Default) // Use heap memory allocator
.ChildOption(ChannelOption.SoKeepalive, true)//Enable TCP heartbeat keepalive mechanism
.ChildOption(ChannelOption.TcpNodelay, true) // Enable TCP Nagle algorithm and disable delayed confirmation
// Used for data processing of a single channel
.ChildHandler(new ActionChannelInitializer<IChannel>(channel =>
{<!-- -->
// Whenever a new connection is received, execute this callback function
IChannelPipeline pipeline = channel.Pipeline;
//Add an idle status handler to detect the idle status of the connection
pipeline.AddLast("IdleChecker", new IdleStateHandler(50, 50, 0));
//Heartbeat packet
//pipeline.AddLast("HeartbeatMessageHandler", new HeartbeatMessage());
//Add custom encoders, decoders and processors
pipeline.AddLast(new TcpServerEncoder(), new TcpServerDecoder(), new TcpServerHandler());
}));
// Bind the specified port and return a channel for server startup
bootstrapChannel = await bootstrap.BindAsync(port);
Console.WriteLine($"Starting the gateway server successfully! Listening port number: {<!-- -->port}");
Console.ReadLine();
}
catch (Exception e)
{<!-- -->
Console.Error.WriteLine(e.Message);
throw new Exception("Failed to start TcpServer! \
" + e.StackTrace);
}
finally
{<!-- -->
}
}
/// <summary>
/// Close TcpServer
/// </summary>
/// <returns></returns>
public static async Task Stop()
{<!-- -->
await Task.WhenAll(
bootstrapChannel.CloseAsync(),
bossGroup.ShutdownGracefullyAsync(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2)),
workerGroup.ShutdownGracefullyAsync(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2))
);
Console.WriteLine("Close the gateway server successfully!");
}
}
}