RabbitMQ (AMQP):
http://localhost:5672
Panel zarządzania RabbitMQ:
http://localhost:15672
Domyślne dane logowania:
Login: admin
Hasło: admin
services:
rabbitmq:
image: rabbitmq:4-management
container_name: rabbitmq
ports:
- "5672:5672" # Port do obsługi protokołu AMQP.
- "15672:15672" # Management plugin UI port. Port interfejsu zarządzania RabbitMQ.
environment:
RABBITMQ_DEFAULT_USER: admin # Domyślna nazwa użytkownika.
RABBITMQ_DEFAULT_PASS: admin # Domyślne hasło użytkownika.
RABBITMQ_DEFAULT_VHOST: my_vhost # Domyślny wirtualny host.
volumes:
- rabbitmq_data:/var/lib/rabbitmq # Persistent storage do przechowywania danych RabbitMQ.
- rabbitmq_logs:/var/log/rabbitmq # Wolumin przechowujący logi RabbitMQ.
volumes:
rabbitmq_data: # Wolumin przechowujący trwałe dane RabbitMQ.
rabbitmq_logs: # Wolumin przechowujący logi serwera RabbitMQ.
docker-compose up
RabbitMQ (AMQP):
http://localhost:5672
Panel zarządzania RabbitMQ:
http://localhost:15672
Domyślne dane logowania:
Login: admin
Hasło: admin
services:
rabbitmq:
image: rabbitmq:4-management
container_name: rabbitmq
ports:
- "5672:5672" # Port protokołu AMQP
- "15672:15672" # Port interfejsu zarządzania
volumes:
- rabbitmq_data:/var/lib/rabbitmq # Trwałość danych
- rabbitmq_logs:/var/log/rabbitmq # Przechowywanie logów
volumes:
rabbitmq_data:
driver: local
rabbitmq_logs:
driver: local
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNetTutorial.01",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
while (true)
{
Console.WriteLine(AppInfo.Value);
Console.WriteLine("Write what you want to send");
Console.WriteLine("Write nothing to exit.");
string usermessage = Console.ReadLine();
if (string.IsNullOrWhiteSpace(usermessage))
break;
string message = usermessage;
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
routingKey: "RabbitMqDotNetTutorial.01",
basicProperties: null, body);
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
}
using RabbitMQ.Client.Events;
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.01",
durable: false,
exclusive: false, autoDelete: false, arguments: null);
Console.WriteLine(AppInfo.Value);
Console.WriteLine(" Waiting for messages.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
Console.Write("-> Received: ");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(message);
Console.ForegroundColor = ConsoleColor.Gray;
//channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
channel.BasicConsume(queue: "RabbitMqDotNet6Tutorial.01",
autoAck: true, consumer: consumer);
char key = 'z';
while (key != 'q' && key != 'Q')
{
Console.WriteLine(" Press [q] or [Q] to exit.");
key = Console.ReadKey().KeyChar;
}
}
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = await factory.CreateConnectionAsync())
using (var channel = await connection.CreateChannelAsync())
{
await channel.QueueDeclareAsync(queue: "R7.01",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
while (true)
{
Console.WriteLine("Write what you want to send");
Console.WriteLine("Write nothing to exit.");
string? usermessage = Console.ReadLine();
if (string.IsNullOrWhiteSpace(usermessage))
break;
string message = usermessage;
var body = Encoding.UTF8.GetBytes(message);
await channel.BasicPublishAsync(exchange: "",
routingKey: "R7.01", body);
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
}
using RabbitMQ.Client.Events;
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = await factory.CreateConnectionAsync())
using (var channel = await connection.CreateChannelAsync())
{
await channel.QueueDeclareAsync(queue: "R7.01", durable: false,
exclusive: false, autoDelete: false, arguments: null);
Console.WriteLine(" Waiting for messages.");
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (model, ea) =>
{
await Task.Run(() =>
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
Console.Write("-> Received: ");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(message);
Console.ForegroundColor = ConsoleColor.Gray;
});
//channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
await channel.BasicConsumeAsync(queue: "R7.01",
autoAck: true, consumer: consumer);
char key = 'z';
while (key != 'q' && key != 'Q')
{
Console.WriteLine(" Press [q] or [Q] to exit.");
key = Console.ReadKey().KeyChar;
}
}
using EasyNetQ;
using (var bus = RabbitHutch.CreateBus
("host=localhost"))
{
for (int i = 0; i < 10; i++)
{
await bus.SendReceive.SendAsync
("1EasyNetMQ.Queue",
new TextMessage()
{
Text = i + ": Hello World from EasyNetQ"
}
);
}
}
using EasyNetQ;
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
await bus.SendReceive.ReceiveAsync("1EasyNetMQ.Queue",
HandleTextMessage);
Console.WriteLine("Listening for messages. Hit to quit.");
Console.ReadLine();
}
static void HandleTextMessage(TextMessage textMessage)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("Got message: {0}", textMessage.Text);
Console.ResetColor();
}
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
//durbale = true ponieważ chcemy aby nasze zdania istniały po restarcie
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.02", true,
false, false, null);
while (true)
{
Console.WriteLine("Write [.] to increase time to do job for worker. Foreach [.]");
Console.WriteLine("Write [!] at least once. To make one worker fail job");
Console.WriteLine("Write [q] or [Q] to exit.");
string whatuserwrote = Console.ReadLine();
if (whatuserwrote == "q" || whatuserwrote == "Q")
break;
if (string.IsNullOrEmpty(whatuserwrote))
continue;
Job job = CreateJob(whatuserwrote);
string message = JsonConvert.SerializeObject(job);
var body = Encoding.UTF8.GetBytes(message);
var properties = channel.CreateBasicProperties();
properties.Persistent = true;
channel.BasicPublish("", routingKey: "RabbitMqDotNet6Tutorial.02",
properties, body);
WriteMessageOnConsole(message);
}
}
void WriteMessageOnConsole(string message)
{
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
Job CreateJob(string usertext)
{
int howManySecondsWillJobTake = usertext.Split('.').Length - 1;
bool shouldFail = usertext.IndexOf('!') > 0;
string message = usertext.Replace("!", "")
.Replace(".", "");
return new Job()
{
Message = message,
HowManySecondsWillJobTake = howManySecondsWillJobTake,
Type = JobType.SendEmail,
ShouldFaillOnWorkerTwo = shouldFail
};
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;
Console.ForegroundColor = GetRandomConsoleColor();
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.02",
durable: true,
exclusive: false, autoDelete: false, arguments: null);
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
Console.WriteLine(" [*] Waiting for jobs.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
byte[] body = ea.Body.ToArray();
var jobAsJsonText = Encoding.UTF8.GetString(body);
Job job = JsonSerializer.Deserialize<Job>(jobAsJsonText);
Console.WriteLine(" [>] Received {0}", job.Message);
Console.WriteLine(" [>] Received {0}", job.Type);
Thread.Sleep(job.HowManySecondsWillJobTake * 1000);
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
};
channel.BasicConsume(queue: "RabbitMqDotNet6Tutorial.02",
autoAck: false,
consumer: consumer);
Console.WriteLine(" Press [enter] to end program.");
Console.ReadLine();
}
ConsoleColor GetRandomConsoleColor()
{
Random _random = new Random();
var consoleColors = Enum.GetValues(typeof(ConsoleColor));
return (ConsoleColor)consoleColors.
GetValue(_random.Next(consoleColors.Length));
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;
Console.ForegroundColor = GetRandomConsoleColor();
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
while (true)
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.02",
durable: true, exclusive: false,
autoDelete: false, arguments: null);
channel.BasicQos(prefetchSize: 0,
prefetchCount: 1,
global: false);
Console.WriteLine(" [*] Waiting for jobs.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
byte[] body = ea.Body.ToArray();
var jobAsJsonText = Encoding.UTF8.GetString(body);
Job job = JsonSerializer.Deserialize<Job>(jobAsJsonText);
Console.WriteLine(" [>] Received {0}", job.Message);
Console.WriteLine(" [>] Received {0}", job.Type);
if (job.ShouldFaillOnWorkerTwo == false)
{
Thread.Sleep(job.HowManySecondsWillJobTake * 500);
channel.BasicAck(deliveryTag: ea.DeliveryTag,
multiple: false);
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
}
else
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" [x] Fail {job.Message}");
Console.WriteLine($" [x] Fail {job.Type}");
Console.ForegroundColor = ConsoleColor.Gray;
}
};
channel.BasicConsume
(queue: "RabbitMqDotNet6Tutorial.02",
autoAck: false,
consumer: consumer);
Console.WriteLine
(" Press [enter] to restart connection.CreateModel()");
Console.ReadLine();
}
}
ConsoleColor GetRandomConsoleColor()
{
Random _random = new Random();
var consoleColors = Enum.GetValues(typeof(ConsoleColor));
return (ConsoleColor)consoleColors.GetValue(_random.Next(consoleColors.Length));
}
using EasyNetQ;
using Newtonsoft.Json;
while (true)
{
Console.WriteLine
("Write [.] to increase time to do job for worker. Foreach [.]");
Console.WriteLine(
"Write [!] at least once. To make one worker fail job");
Console.WriteLine
("Write [q] or [Q] to exit.");
string whatuserwrote = Console.ReadLine();
if (whatuserwrote == "q" || whatuserwrote == "Q")
break;
if (string.IsNullOrEmpty(whatuserwrote))
continue;
Job job = CreateJob(whatuserwrote);
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
await bus.SendReceive.SendAsync<Job>("2EasyNetMQ.NewJob.Queue"
, job
);
}
string message = JsonConvert.SerializeObject(job);
WriteMessageOnConsole(message);
}
Job CreateJob(string usertext)
{
int howManySecondsWillJobTake = usertext.Split('.').Length - 1;
bool shouldFail = usertext.IndexOf('!') > 0;
string message = usertext.Replace("!", "")
.Replace(".", "");
return new Job()
{
Message = message,
HowManySecondsWillJobTake = howManySecondsWillJobTake,
Type = JobType.SendEmail,
ShouldFaillOnWorkerTwo = shouldFail
};
}
void WriteMessageOnConsole(string message)
{
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
using EasyNetQ;
var connectionString = "host=localhost";
// RabbitMQ connection string
using (var bus = RabbitHutch.CreateBus(connectionString))
{
var queueName = "2EasyNetMQ.NewJob.Queue";
// Replace with your queue name
bus.Advanced.QueueDeclare(queueName, durable: true, false, false);
await bus.SendReceive.ReceiveAsync
(queueName, HandleMessage, x => x.WithAutoDelete(false));
Console.WriteLine("Press [enter] to exit.");
Console.ReadLine();
}
static void HandleMessage(Job job)
{
try
{
// Process the message
Console.WriteLine(" [>] Received {0}", job.Message);
Console.WriteLine(" [>] Received {0}", job.Type);
Thread.Sleep(job.HowManySecondsWillJobTake * 1000);
// Manually ACK the message
// YOU CAN'T DO THIS WITH EasyNetQ
// Note: EasyNetQ automatically ACKs messages by default.
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
}
catch (Exception ex)
{
Console.WriteLine($"Error processing message: {ex.Message}");
// Handle any exceptions or log errors here
}
}
using EasyNetQ;
var connectionString = "host=localhost";
// RabbitMQ connection string
using (var bus = RabbitHutch.CreateBus(connectionString))
{
var queueName = "2EasyNetMQ.NewJob.Queue";
// Replace with your queue name
bus.Advanced.QueueDeclare(queueName, durable: true, false, false);
// Subscribe to messages
await bus.SendReceive.ReceiveAsync
(queueName, HandleMessage, x => x.WithAutoDelete(false));
Console.WriteLine("Press [enter] to exit.");
Console.ReadLine();
}
static void HandleMessage(Job job)
{
if (job.ShouldFaillOnWorkerTwo == false)
{
// Note: EasyNetQ automatically ACKs messages by default.
Thread.Sleep(job.HowManySecondsWillJobTake * 500);
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
}
else
{
// Manually N-ACK the message
// YOU CAN'T DO THIS WITH EasyNetQ
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" [x] Fail {job.Message}");
Console.WriteLine($" [x] Fail {job.Type}");
Console.ForegroundColor = ConsoleColor.Gray;
// THIS WILL SEND MESSAGE TO DEAD-LETTER QUEQUE
// DEFAULT NAME OF THIS QUEQUE IS
// EasyNetQ_Default_Error_Queue
throw new Exception("JOB FAILED");
}
}
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "RabbitMqDotNet6Tutorial.03",
type: ExchangeType.Fanout,
durable: true, autoDelete: false, arguments: null);
while (true)
{
var message = "Daj mi ocene 5!";
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "RabbitMqDotNet6Tutorial.03",
routingKey: "",
basicProperties: null, body: body);
Console.WriteLine(" [x] Sent {0}", message);
Console.ReadLine();
}
}
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "RabbitMqDotNet6Tutorial.03",
type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: queueName,
exchange:
"RabbitMqDotNet6Tutorial.03", routingKey: "");
Console.WriteLine(" [*] Waiting for logs.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
byte[] body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] {0}", message);
};
channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
using EasyNetQ;
using EasyNetQ.Topology;
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
var message = "Daj mi ocene 5!";
await bus.PubSub.PublishAsync<String>(
message
);
Console.WriteLine(" [x] Sent {0}", message);
Console.ReadLine();
}
static void AddCustom(IBus bus)
{
var exchange = bus.Advanced.ExchangeDeclare(
"3EasyNetQ.PublishToMany.Exchange", ExchangeType.Fanout);
var queue = bus.Advanced.QueueDeclare
("3EasyNetQ.PublishToMany.Queue",
durable: true, exclusive: false, autoDelete: true);
bus.Advanced.Bind(exchange, queue, "#");
}
using EasyNetQ;
using EasyNetQ.Topology;
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
AddCustom(bus);
await bus.PubSub.SubscribeAsync("3EasyNetQ.SubscribeSecond", HandleTextMessage);
Console.WriteLine("Listening for messages. Hit to quit.");
Console.ReadLine();
}
static void HandleTextMessage(string textMessage)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("Got message: {0}", textMessage);
Console.ResetColor();
}
static void AddCustom(IBus bus)
{
var exchange = bus.Advanced.ExchangeDeclare(
"3EasyNetQ.PublishToMany.Exchange", ExchangeType.Fanout);
var queue = bus.Advanced.QueueDeclare
("3EasyNetQ.PublishToMany.Queue",
durable: true, exclusive: false, autoDelete: true);
bus.Advanced.Bind(exchange, queue, "#");
}
public class DocumentSavedEvent
{
private Guid id;
private string filename;
[JsonConstructor]
public DocumentSavedEvent(Guid id, string filename)
{
this.Id = id;
this.FileName = filename;
}
public Guid Id { get => id; set => id = value; }
public string FileName { get => filename; set => filename = value; }
}
public class TradeRecordedEvent
{
private Guid id;
private string commodity;
private int quantity;
[JsonConstructor]
public TradeRecordedEvent(Guid id, string commodity, int quantity)
{
this.Id = id;
this.Commodity = commodity;
this.Quantity = quantity;
}
public Guid Id
{ get => id; set => id = value; }
public string Commodity
{ get => commodity; set => commodity = value; }
public int Quantity
{ get => quantity; set => quantity = value; }
}
public class UserLoggedEvent
{
private Guid id;
private string username;
[JsonConstructor]
public UserLoggedEvent(Guid id, string username)
{
this.Id = id;
this.UserName = username;
}
public Guid Id { get => id; set => id = value; }
public string UserName { get => username; set => username = value; }
}
using Rebus.Activation;
using Rebus.Bus;
using Rebus.Config;
using (var activator = new BuiltinHandlerActivator())
{
var bus = Configure.With(activator)
.Transport(t => t.UseRabbitMqAsOneWayClient
("amqp://guest:guest@localhost:5672"))
.Start();
await Task.Delay(500);
Console.WriteLine("Trading is running");
while (true)
{
Console.WriteLine(@"
a) Publish TradeRecordedEvent
b) Publish DocumentSavedEvent
c) Publish UserLogedEvent
q) Quit");
var keyChar = char.ToLower(Console.ReadKey(true).KeyChar);
switch (keyChar)
{
case 'a':
await SendTradeRecordedEventAsync(bus);
break;
case 'b':
await SendDocumentSavedAsync(bus);
break;
case 'c':
await SendUserLoggedEventAsync(bus);
break;
case 'q':
break;
default:
Console.WriteLine("There's no option ({0})", keyChar);
break;
}
}
}
async Task SendTradeRecordedEventAsync(IBus bus)
{
Console.WriteLine("Please enter new trade details");
Console.WriteLine(" commodity > ");
var commodity = Console.ReadLine();
if (string.IsNullOrWhiteSpace(commodity)) return;
int quantity;
Console.WriteLine(" quantity > ");
while (!int.TryParse(Console.ReadLine(), out quantity)) ;
var tradeEvent = new TradeRecordedEvent
(Guid.NewGuid(), commodity, quantity);
await bus.Publish(tradeEvent);
Console.WriteLine("Sended tradeEvent");
}
async Task SendDocumentSavedAsync(IBus bus)
{
Console.WriteLine("Please enter new document details");
Console.WriteLine(" document name > ");
var documentName = Console.ReadLine();
if (string.IsNullOrWhiteSpace(documentName)) return;
var docEvent = new DocumentSavedEvent
(Guid.NewGuid(), documentName);
await bus.Publish(docEvent);
Console.WriteLine("Sended docEvent");
}
async Task SendUserLoggedEventAsync(IBus bus)
{
Console.WriteLine("Please enter new trade details");
Console.WriteLine(" Username > ");
var username = Console.ReadLine();
if (string.IsNullOrWhiteSpace(username)) return;
var userEvent = new UserLoggedEvent(Guid.NewGuid(), username);
await bus.Publish(userEvent);
Console.WriteLine("Sended userEvent");
}
using Rebus.Activation;
using Rebus.Config;
using Rebus.Routing.TypeBased;
using (var activator = new BuiltinHandlerActivator())
{
activator.Register(() => new TradeRecordedEventHandler());
activator.Register(() => new UserLoggedEventHandler());
activator.Register(() => new DocumentSavedEventHandler());
var bus = Configure.With(activator)
.Logging
(l => l.ColoredConsole(minLevel: Rebus.Logging.LogLevel.Warn))
.Transport
(t => t.UseRabbitMq("amqp://guest:guest@localhost:5672",
"Rebus.SystemTooBig.Trading"))
.Routing
(r => r.TypeBased().
MapAssemblyOf<TradeRecordedEvent>("Rebus.SystemTooBig.Trading"))
.Start();
await bus.Subscribe<TradeRecordedEvent>();
await bus.Subscribe<UserLoggedEvent>();
await bus.Subscribe<DocumentSavedEvent>();
Console.WriteLine("Invoicing is running - press Enter to quite");
Console.ReadKey();
}
internal class DocumentSavedEventHandler : IHandleMessages<DocumentSavedEvent>
{
public DocumentSavedEventHandler()
{
}
public async Task Handle(DocumentSavedEvent message)
{
Console.WriteLine
($"Document was saved: {message.Id} ({message.FileName})");
}
}
internal class TradeRecordedEventHandler : IHandleMessages<TradeRecordedEvent>
{
public TradeRecordedEventHandler()
{
}
public async Task Handle(TradeRecordedEvent message)
{
Console.WriteLine
($"Invocing trade: {message.Id} ({message.Quantity} x {message.Commodity})");
}
}
internal class UserLoggedEventHandler : IHandleMessages<UserLoggedEvent>
{
public UserLoggedEventHandler()
{
}
public async Task Handle(UserLoggedEvent message)
{
Console.WriteLine
($"User was logged: {message.Id} ({message.UserName})");
}
}
WHAT
WHY
WINDOWS 11
WHAT IS WMIC
services:
zookeeper: #Zarządza metadanymi klastru Kafka i koordynuje pracę brokerów.
image: bitnami/zookeeper:3.9.1
# to survive the container restart
tmpfs: "/zktmp"
environment:
ALLOW_ANONYMOUS_LOGIN: 'yes' #Zezwala na logowanie bez uwierzytelnienia.
ports:
- "2181:2181"
kafka1:
image: bitnami/kafka:3.7.0
depends_on:
- zookeeper
environment:
KAFKA_BROKER_ID: 1 #Identyfikator brokera.
KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181 #Adres Zookeepera.
KAFKA_CFG_LISTENERS: INTERNAL://:9092,EXTERNAL://0.0.0.0:19092
KAFKA_CFG_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092,EXTERNAL://localhost:19092
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
# opcjonalne
KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true' #Automatyczne tworzenie topików.
ALLOW_PLAINTEXT_LISTENER: 'yes' #Zezwala na komunikację bez szyfrowania.
ports:
- "9092:9092" #Port wewnętrzny brokera.
- "19092:19092" #Port zewnętrzny dla lokalnego dostępu.
volumes:
- kafka_data1:/bitnami/kafka #Wolumen do przechowywania danych brokera.
kafka2:
image: bitnami/kafka:3.7.0
depends_on:
- zookeeper
environment:
KAFKA_BROKER_ID: 2
KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_CFG_LISTENERS: INTERNAL://:9093,EXTERNAL://0.0.0.0:19093
KAFKA_CFG_ADVERTISED_LISTENERS: INTERNAL://kafka2:9093,EXTERNAL://localhost:19093
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
# optional - enable topic auto create
KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true'
ALLOW_PLAINTEXT_LISTENER: 'yes'
ports:
- "9093:9093"
- "19093:19093"
volumes:
- kafka_data2:/bitnami/kafka
kafka3:
image: bitnami/kafka:3.7.0
depends_on:
- zookeeper
environment:
KAFKA_BROKER_ID: 3
KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_CFG_LISTENERS: INTERNAL://:9094,EXTERNAL://0.0.0.0:19094
KAFKA_CFG_ADVERTISED_LISTENERS: INTERNAL://kafka3:9094,EXTERNAL://localhost:19094
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
# optional - enable topic auto create
KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true'
ALLOW_PLAINTEXT_LISTENER: 'yes'
ports:
- "9094:9094"
- "19094:19094"
volumes:
- kafka_data3:/bitnami/kafka
kafka-ui: #Interfejs webowy do monitorowania i zarządzania klastrem Kafka.
image: provectuslabs/kafka-ui:latest
depends_on:
- kafka1
- kafka2
- kafka3
ports:
- "1111:8080" #Dostępne na porcie `1111` na hoście.
environment:
KAFKA_CLUSTERS_0_NAME: local
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka1:9092,kafka2:9093,kafka3:9094
KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
volumes:
kafka_data1:
driver: local
kafka_data2:
driver: local
kafka_data3:
driver: local
## **Jak to działa**
1. **Zookeeper**:
- Uruchamia się jako pierwszy i koordynuje pracę brokerów.
2. **Kafka Brokers**:
- Trzy brokery komunikują się z Zookeeperem i między sobą.
- Każdy broker obsługuje Inne porty, co pozwala na równoległą pracę w klastrze.
- Ich adresy `http://localhost:19092`
`http://localhost:19093`
`http://localhost:19094`
3. **Kafka UI**:
- Dostępny przez przeglądarkę pod adresem `http://localhost:1111`.
- Umożliwia monitorowanie i zarządzanie klastrem w prosty sposób.
---
## **Uwagi**
- Zmienne środowiskowe w konfiguracji pozwalają na dużą elastyczność i dostosowanie
klastra do różnych potrzeb.
- Porty **INTERNAL** są używane wewnętrznie przez brokerów, natomiast porty **EXTERNAL**
są wystawiane na zewnątrz, aby aplikacje mogły łączyć się z Kafką.
using _1Kafka.Confluent.Common;
using Confluent.Kafka;
using System.Net;
using System.Text.Json;
string bootstrapServers = "localhost:9092";
string topic = "1Kafka.Confluent.Test";
char c = ' ';
int messageId = 0;
while (c != 'q')
{
var messagerequest = GetMessageRequest();
string json = JsonSerializer.Serialize(messagerequest);
await SendMessageRequest(topic, json);
Console.WriteLine("Press any key to continue");
Console.WriteLine("Press 'q' to exit");
c = Console.ReadKey().KeyChar;
Console.WriteLine("");
}
MessageRequest GetMessageRequest()
{
string message = "";
int color = -1;
while (string.IsNullOrWhiteSpace(message) || color == -1)
{
Console.WriteLine("Write Your Message");
message = Console.ReadLine();
Console.WriteLine("Choose Color");
for (int i = 0; i < 16; i++)
{
Console.ForegroundColor = (ConsoleColor)i;
Console.WriteLine($"{i} . {((ConsoleColor)i).ToString()}");
}
Console.ResetColor();
int colora;
bool check = int.TryParse(Console.ReadLine(), out colora);
if (!check)
continue;
color = colora;
}
messageId++;
MessageRequest messageRequest = new()
{
Color = color,
Message = message,
Id = messageId,
UniqueId = Guid.NewGuid(),
StartProducer = DateTime.Now
};
return messageRequest;
}
async Task<bool> SendMessageRequest(string topic, string message)
{
ProducerConfig config = new ProducerConfig
{
BootstrapServers = bootstrapServers,
ClientId = Dns.GetHostName(),
};
try
{
using (var producer = new ProducerBuilder
<Null, string>(config).Build())
{
var result = await producer.ProduceAsync
(topic, new Message<Null, string>
{
Value = message
});
Console.WriteLine($"Delivery Timestamp:{result.Timestamp.UtcDateTime}");
return await Task.FromResult(true);
}
}
catch (Exception ex)
{
Console.WriteLine($"Error occured: {ex.Message}");
}
return await Task.FromResult(false);
}
using _1Kafka.Confluent.Common;
using Confluent.Kafka;
using System.Net;
using System.Text.Json;
int unique = 5;
string topic = "1Kafka.Confluent.Test";
string groupId = "1Kafka.Confluent.Test_Group_" + unique.ToString();
string bootstrapServers = "localhost:9092";
Console.WriteLine($"{topic} GROUP ID : {groupId}");
var config = new ConsumerConfig
{
GroupId = groupId,
BootstrapServers = bootstrapServers,
AutoOffsetReset = AutoOffsetReset.Earliest
};
try
{
using (var consumerBuilder = new ConsumerBuilder
<Ignore, string>(config).Build())
{
consumerBuilder.Subscribe(topic);
var cancelToken = new CancellationTokenSource();
try
{
while (true)
{
var consumer = consumerBuilder.Consume
(cancelToken.Token);
var messageRequest = JsonSerializer.Deserialize
<MessageRequest>
(consumer.Message.Value);
Console.WriteLine($"\nProducerStarted: {messageRequest.StartProducer}");
Console.WriteLine($"\nProcessing Number: {messageRequest.Id}");
Console.WriteLine($"Processing UniqueId: {messageRequest.UniqueId}");
if (messageRequest.Color >= 0 && messageRequest.Color <= 15)
Console.ForegroundColor = (ConsoleColor)messageRequest.Color;
Console.WriteLine(messageRequest.Message);
Console.ResetColor();
Console.WriteLine("");
}
}
catch (OperationCanceledException)
{
consumerBuilder.Close();
}
}
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
using KafkaFlow;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddKafka(
kafka => kafka
.AddCluster(cluster =>
{
const string topicName = "1KafkaFlow";
cluster
.WithBrokers(new[] { "localhost:9092" })
.CreateTopicIfNotExists(topicName,
numberOfPartitions: 3, replicationFactor: 3)
.AddProducer(
name: "1KafkaFlow.Producer",
producer => producer
.DefaultTopic(topicName)
.AddMiddlewares(middlewares =>
middlewares
.AddSerializer<JsonCoreSerializer>()));
})
);
var app = builder.Build();
app.MapPost("/add", RequestHandler.HandleAsync);
app.Run();
// Handler
public static class RequestHandler
{
public static async Task<IResult> HandleAsync(
IProducerAccessor producerAccessor,
AddTaskRequest request, CancellationToken cancellationToken)
{
var producer = producerAccessor.GetProducer("1KafkaFlow.Producer");
await producer.ProduceAsync(
null,
request
);
return Results.Accepted();
}
}
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
const string topicName = "1KafkaFlow";
int unique = 5;
string groupId = "1KafkaFlow_Group_" + unique.ToString();
var services = new ServiceCollection();
services.AddKafkaFlowHostedService(
kafka => kafka
.UseMicrosoftLog()
.AddCluster(cluster =>
{
cluster
.WithBrokers(new[] { "localhost:19092" })
.AddConsumer(consumer =>
consumer
.Topic(topicName)
.WithGroupId(groupId)
.WithBufferSize(100)
.WithWorkersCount(3)
.WithAutoOffsetReset(KafkaFlow.AutoOffsetReset.Earliest)
.AddMiddlewares(middlewares => middlewares
.AddDeserializer<JsonCoreDeserializer>()
.AddTypedHandlers(handlers =>
handlers.AddHandler<AddTaskHandler>()
)
)
);
})
);
services.AddLogging(configure => configure.AddConsole());
var provider = services.BuildServiceProvider();
var bus = provider.CreateKafkaBus();
await bus.StartAsync();
using KafkaFlow;
using Microsoft.Extensions.Logging;
public class AddTaskHandler : IMessageHandler<AddTaskRequest>
{
private readonly ILogger<AddTaskHandler> _logger;
public AddTaskHandler(ILogger<AddTaskHandler> logger)
{
_logger = logger;
}
public Task Handle(IMessageContext context, AddTaskRequest message)
{
if (message.DueDate.HasValue)
_logger.LogInformation("New Task {Title} scheduled to {DueDate}",
message.Title,
message.DueDate);
return Task.CompletedTask;
}
}
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
const string topicName = "1KafkaFlow";
int unique = 6;
string groupId = "1KafkaFlow_Group_" + unique.ToString();
var services = new ServiceCollection();
services.AddLogging(configure => configure.AddConsole());
services.AddKafkaFlowHostedService(
kafka => kafka
.UseMicrosoftLog()
.AddCluster(cluster =>
{
cluster
.WithBrokers(new[] { "localhost:19092" })
.AddConsumer(consumer =>
consumer
.Topic(topicName)
.WithGroupId(groupId)
.WithBufferSize(100)
.WithWorkersCount(3)
.WithAutoOffsetReset(KafkaFlow.AutoOffsetReset.Earliest)
.AddMiddlewares(middlewares => middlewares
.Add<CatchErrorsMiddleware>()
.AddDeserializer<JsonCoreDeserializer>()
.AddBatching(5, TimeSpan.FromSeconds(5))
.Add<StatisticsMiddleware>()
.Add<ReadMiddleware>()
)
);
})
);
var provider = services.BuildServiceProvider();
var bus = provider.CreateKafkaBus();
await bus.StartAsync();
Console.WriteLine("Press key to exit");
Console.ReadKey();
public class CatchErrorsMiddleware : IMessageMiddleware
{
public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
{
try
{
await next(context);
}
catch (Exception ex)
{
Console.WriteLine($"{ex}");
}
}
}
public class StatisticsMiddleware : IMessageMiddleware
{
private static int _total = 0;
public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
{
var batch = context.GetMessagesBatch();
_total += batch.Count;
Console.WriteLine($"Current Total: {_total}");
await next(context);
}
}
public class ReadMiddleware : IMessageMiddleware
{
public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
{
var bytes = context.Message.Value as byte[];
var message = Encoding.UTF8.GetString(bytes);
Console.WriteLine($"Read message: {message}");
await next(context);
}
}
var builder = WebApplication.CreateBuilder(args);
builder.Services.Configure<KafkaConfig>(builder.Configuration.GetSection("KafkaConfig"));
builder.Logging.AddConsole();
builder.Logging.SetMinimumLevel(LogLevel.Debug);
builder.Services.AddLogging(configure => configure.AddConsole());
var server = builder.Configuration["KafkaConfig:BootstrapServers"];
var topic = builder.Configuration["KafkaConfig:Topic"];
var groupid = builder.Configuration["KafkaConfig:GroupId"];
builder.Services
.AddKafka(kafka => kafka
.AddCluster(cluster => cluster
.WithBrokers(new[] { server })
.AddConsumer(consumer => consumer
.Topic(topic)
.WithGroupId(groupid)
.WithWorkersCount(1)
.WithBufferSize(10)
)
.EnableTelemetry("kafka-flow.admin")
// you can use the same topic used in EnableAdminMessages, if need it
.EnableAdminMessages(
"kafka-flow.admin" // the admin topic
)
))
.AddControllers();
builder.Services
.AddSwaggerGen(
c =>
{
c.SwaggerDoc(
"kafka-flow",
new OpenApiInfo
{
Title = "KafkaFlow Admin",
Version = "kafka-flow",
});
});
var app = builder.Build();
app.MapControllers();
app.UseKafkaFlowDashboard();
app.UseSwagger();
app.UseSwaggerUI(c =>
{
c.SwaggerEndpoint("/swagger/kafka-flow/swagger.json", "KafkaFlow Admin");
});
var kafkaBus = app.Services.CreateKafkaBus();
await kafkaBus.StartAsync();
await app.RunAsync();
using KafkaFlow;
using Microsoft.Extensions.DependencyInjection;
services.AddKafka(kafka => kafka
.AddCluster(cluster => cluster
.WithBrokers(new[] { "localhost:9092" })
.AddConsumer(consumer => consumer
.Topic("topic-name")
.WithGroupId("sample-group")
.WithAutoOffsetReset(AutoOffsetReset.Earliest)
...
)
)
);
//Earliest: Najwcześniej: Odczytywane od początku tematu.
//Latest: Najnowsze: Odczytuje tylko nowe wiadomości.
//Acks.None acks=0
//Acks.Leader acks=1
//Acks.All acks=all
using KafkaFlow;
using KafkaFlow.Producers;
using Microsoft.Extensions.DependencyInjection;
services.AddKafka(kafka => kafka
.AddCluster(cluster => cluster
.WithBrokers(new[] { "localhost:9092" })
.AddProducer(
"product-events",
producer =>
producer
.WithAcks(Acks.Leader)
)
)
);
//acks=0 Jeśli ustawione na zero, producent nie będzie czekał na żadne potwierdzenie
// z serwera.Rekord zostanie natychmiast dodany do bufora gniazda i uznany za wysłany.
// Nie można zagwarantować, że serwer otrzymał rekord w tym przypadku, a konfiguracja
// ponawiania prób nie będzie działać (ponieważ klient zazwyczaj nie będzie wiedział o
// żadnych awariach). Offset zwracany dla każdego rekordu będzie zawsze ustawiony na -1.
//acks=1 Oznacza to, że lider zapisze rekord do swojego lokalnego dziennika, ale odpowie bez
// oczekiwania na pełne potwierdzenie od wszystkich obserwujących.W takim przypadku, jeśli
// lider ulegnie awarii natychmiast po potwierdzeniu rekordu, ale przed jego replikacją przez
// podwładnych, rekord zostanie utracony.
//acks= all Oznacza to, że lider będzie czekał na potwierdzenie rekordu przez cały zestaw
// zsynchronizowanych replik. Gwarantuje to, że rekord nie zostanie utracony, dopóki
// co najmniej jedna zsynchronizowana replika pozostanie przy życiu.Jest to najsilniejsza
// dostępna gwarancja.Jest to równoważne ustawieniu acks= -1.