Initial RabbitMQ Tutorial commit (#288)

This commit is contained in:
Dave Tillman 2023-05-09 06:38:59 -06:00 коммит произвёл GitHub
Родитель d2768b700c
Коммит 96f7dac642
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
107 изменённых файлов: 4099 добавлений и 0 удалений

Просмотреть файл

@ -0,0 +1,379 @@
# RabbitMQ Tutorial - "Hello World!"
## Introduction
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
RabbitMQ is a message **broker**; it accepts and forwards messages.
You can think of it as a post office; when you put the mail that you want sent in a post office box,
you can be sure that the letter carrier will eventually deliver the mail to your recipient.
In this analogy, RabbitMQ is a post office box, a post office, and a letter carrier.
The major difference between RabbitMQ and the post office is that it doesn't deal with paper,
instead it accepts, stores, and forwards binary blobs of data **messages**.
RabbitMQ, and messaging in general, use some jargon as follows:
- **Producing** means nothing more than sending a message. A program that sends messages is a **producer**. In these tutorials we use the symbol below to represent a **producer**.
<p>
<img src="../img/tutorials/producer.png">
</p>
- **A queue** is the name for the post office box in RabbitMQ. Although messages flow through RabbitMQ and your applications, they can only be stored inside a **queue**.
A **queue** is only bound by the host's memory &amp; disk limits, it's essentially a large message buffer.
Many **producers** can send messages that go to one queue, and many **consumers** can try to receive data from a single **queue**. We use the symbol below to represent a **queue**.
<p>
<img src="../img/tutorials/queue.png">
</p>
- **Consuming** has a similar meaning to receiving a message. A **consumer** is a program that mostly waits to receive messages. We use the symbol below to represent a **consumer**
<p>
<img src="../img/tutorials/consumer.png">
</p>
Note that the **producer**, **consumer**, and **broker** do not have to reside on the same host; indeed in most applications they don't.
An application can be both a **producer** and **consumer**, at the same time.
## "Hello World" (using Steeltoe)
In this part of the tutorial we'll write two programs using the Steeltoe Messaging framework;
a **producer** that sends a single message, and a **consumer** that receives
messages and prints them out. We'll gloss over some of the details in
the Steeltoe API, concentrating on this very simple thing just to get
started. It's a "Hello World" of messaging.
In the diagram below, "P" is our **producer** and "C" is our **consumer**. The
box in the middle is a queue - a message buffer that RabbitMQ keeps
on behalf of the **consumer**.
<p>
<img src="../img/tutorials/python-one.png">
</p>
### The Steeltoe Messaging Framework
RabbitMQ speaks multiple protocols and message formats. This tutorial and the others in this series use AMQP 0-9-1, which is an open, general-purpose protocol for messaging.
There are a number of different clients for RabbitMQ supporting
[many different languages and libraries](http://rabbitmq.com/devtools.html).
In this tutorial, we'll be using .NET Core and the C# language. In addition we will be using the Steeltoe
Messaging library to help simplify writing messaging applications in .NET.
We have also chosen to use Visual Studio 2022 to edit and build the project; but we could have chosen VSCode as well.
The [source code of the project](https://github.com/steeltoeoss/samples/tree/main/messaging/tutorials)
is available online. You can either just run the finished tutorials or you can do the tutorials from scratch by following the steps outlined in each of tutorials writeup.
If you choose to start from scratch, open Visual Studio and create a new **Console** application using the VS2022 template:
<p>
<img src="../img/tutorials/VS2022NewConsoleApp.png" alt="New ConsoleApp"/>
</p>
Name the project `Receiver` and select a directory location such as `c:\\workspace\\Tutorials`.
Choose a solution name of Tutorial1 and uncheck the `Place solution and project in the same directory` as you will be adding another project to this solution next.
<p>
<img src="../img/tutorials/VS2022NewConsoleAppConfigureProject.png" alt="Configure Project"/>
</p>
Next add another project to the solution. Choose a **Worker Service** project type this time:
<p>
<img src="../img/tutorials/VS2022NewWorkService.png" alt="New Worker Service"/>
</p>
Name this project `Sender` and select the same directory location and solution name you picked earlier.
<p>
<img src="../img/tutorials/VS2022NewWorkServiceConfigure.png" alt="New Worker Service Configure"/>
</p>
When you are done with the above, add a new class to the `Receiver` project.
Name this class `Tut1Receiver`; this will be the class we use to receive messages from the sender.
Next, in the `Sender` project, rename the `Worker.cs` file to `Tut1Sender.cs`.
Finally, in both of the project `.csproj` files add the Steeltoe RabbitMQ Messaging package reference:
```xml
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
```
After these changes your solution should look something like the following:
<p>
<img src="../img/tutorials/VS2022Solution.png" alt="New Worker Service Configure"/>
</p>
## Configuring the Projects
Steeltoe Messaging offers numerous features you can use to tailor your messaging application, but in this tutorial we only highlight a few that help us get our application up and running with a minimal amount of code.
First, Steeltoe RabbitMQ Messaging applications have the option of using the `RabbitMQHost` to setup and configure the .NET `Host` used to run the application. The `RabbitMQHost` is a simple host that is configured and behaves just like the [.NET Core Generic Host](https://learn.microsoft.com/en-us/dotnet/core/extensions/generic-host) but also configures the service container with all the Steeltoe components required to send and receive messages with RabbitMQ.
Specifically it adds and configures the following services:
- `RabbitTemplate` - used to send (i.e. **producer**) and receive (i.e. **consumer**) messages.
- `RabbitAdmin` - used to administer (i.e. create, delete, update, etc.) RabbitMQ entities (i.e. Queues, Exchanges, Bindings, etc.). At startup the the RabbitAdmin looks for any RabbitMQ entities defined in the service container and attempts to define them in the broker.
- `RabbitListener Attribute processor` - processes all `RabbitListener` attributes and creates RabbitContainers (i.e.**consumers**) for each listener.
- `Rabbit Container Factory` - a component used to create and manage all the RabbitContainers (i.e.**consumers**) in the application
- `Rabbit Message Converter` - a component used to translate .NET objects to a byte stream to be sent and received. Defaults to .NET serialization, but can be easily changed to use `json`.
- `Caching Connection Factory` - used to create and cache connections to the RabbitMQ broker. All of the above components use it when interacting with the broker. By default it is configured to use `localhost` and port (`5672`).
Throughout the tutorials we will explain how all the above components come into play when building and running a messaging application.
To get started lets change the `Program.cs` file for the `Receiver` project. Specifically, lets use the `RabbitMQHost.CreateDefaultBuilder(args)` method to create a RabbitMQ host.
```csharp
// Create a default RabbitMQ host builder
RabbitMQHost.CreateDefaultBuilder(args)
```
Next use the `.ConfigureServices()` method on the builder to further configure the services in the host.
First use the Steeltoe extension method `.AddRabbitQueue(...)` to configure a `Queue` in the service container. We do this so that the `RabbitAdmin` that has been added to the service container for you will find it and at startup use it to create and configure the queue for us on the broker when the application starts up.
```csharp
// Add queue to service container to be declared at startup
services.AddRabbitQueue(new Queue(QueueName));
```
Next configure `Tut1Receiver`. This is the component that will process messages received on the queue we configured above. This is done by adding `Tut1Receiver` as a singleton in the service container and then also configuring Steeltoe messaging to recognize the class as a `RabbitListener`. As a result, in the background, the Steeltoe `RabbitListener Attribute processor` and the `Rabbit Container factory` mentioned above use this information and more to create a `RabbitContainer` (i.e. **consumer**) that consumes messages from the queue and invokes methods in the class (e.g. `Tut1Receiver`) to process it.
Note, at this point we have not explained how to tie together a queue, `Tut1Receiver` and the method; that comes next.
```csharp
// Add the rabbit listener component
services.AddSingleton<Tut1Receiver>();
// Tell Steeltoe the component is a listener
services.AddRabbitListeners<Tut1Receiver>();
```
When your done, the `Program.cs` file for the `Receiver` project looks as follows:
```csharp
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
internal const string QueueName = "hello";
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
// Add the rabbit listener
services.AddSingleton<Tut1Receiver>();
services.AddRabbitListeners<Tut1Receiver>();
})
.Build()
.Run();
}
}
}
```
Next we'll change the `Program.cs` file for the `Sender` project.
Use the `RabbitMQHost.CreateDefaultBuilder(args)` method as well to create a RabbitMQ host in the Sender. Also add the `Queue` into the service container so it gets declared in the broker. This allows us to start either the sender or the receiver and regardless of which one starts first, the queue gets declared in the broker.
With these changes done, the `Program.cs` file for the `Sender` project looks as follows:
```csharp
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
// The name of the queue that will be created
internal const string QueueName = "hello";
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
services.AddHostedService<Tut1Sender>();
})
.Build()
.Run();
}
}
}
```
## Sending
<div class="diagram">
<img src="../img/tutorials/sending.png" alt="sending" />
</div>
Now there is very little code that needs to go into the
sender and receiver classes. The sender leverages the Steeltoe`RabbitTemplate` that the `RabbitMQHost` adds to the service container for you which you can use to send messages. We will inject it into the sender by adding it to the constructor of `Tut1Sender`.
Here is the code for the sender:
```csharp
using Steeltoe.Messaging.RabbitMQ.Core;
namespace Sender
{
public class Tut1Sender : BackgroundService
{
private readonly ILogger<Tut1Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
public Tut1Sender(ILogger<Tut1Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
await _rabbitTemplate.ConvertAndSendAsync(Program.QueueName, "Hello World!");
_logger.LogInformation("Worker running at: {time}, sent message!", DateTimeOffset.Now);
await Task.Delay(1000, stoppingToken);
}
}
}
}
```
You'll notice that Steeltoe removes the typical boilerplate .NET code needed to send and receive messages
leaving you with only the logic of the messaging to be concerned
about. Steeltoe wraps the boilerplate RabbitMQ client classes with
a `RabbitTemplate` that can be easily injected into the sender. The template has been
pre-configured with a connection to the broker using the `Caching Connection Factory` mentioned earlier.
All that is left is to create a message and invoke the template's
`ConvertAndSend***()` method passing in the queue name that
we defined and the message we wish to send.
> #### Sending doesn't work!
>
> If this is your first time using RabbitMQ and you don't see the "Sent"
> message then you may be left scratching your head wondering what could
> be wrong. Maybe the broker was started without enough free disk space
> (by default it needs at least 200 MB free) and is therefore refusing to
> accept messages. Check the broker log file to confirm and reduce the
> limit if necessary. The <a
> href="https://www.rabbitmq.com/configure.html#config-items">configuration
> file documentation</a> will show you how to set <code>disk_free_limit</code>.
## Receiving
The receiver is equally simple. We annotate our receiver
class with `RabbitListener` attribute and pass in the name of the queue to the attribute. This ties the method to the queue such that all messages that arrive on the queue will be delivered to the method.
In this case we will annotate a `void Receive(string input)` method which has a parameter (i.e. `input`) indicates the type of object from the message payload we expect to receive from the queue. In our tutorial we will be sending and receiving strings via the queue. Behind the scenes, Steeltoe will use the `Rabbit Message converter` mentioned earlier to convert the incoming message payload to the type you defined in the `Receive()` method.
Here is the code for the receiver:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
namespace Receiver
{
public class Tut1Receiver
{
private readonly ILogger _logger;
public Tut1Receiver(ILogger<Tut1Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = Program.QueueName)]
public void Receive(string input)
{
_logger.LogInformation($"Received: {input}");
}
}
}
```
## Putting it all together
We must now build the solution.
```bash
cd tutorials\tutorial1
dotnet build
```
To run the receiver, execute the following commands:
```bash
# receiver
cd receiver
dotnet run
```
Open another shell to run the sender:
```bash
# sender
cd sender
dotnet run
```
> #### Listing queues
>
> You may wish to see what queues RabbitMQ has and how many
> messages are in them. You can do it (as a privileged user) using the `rabbitmqctl` CLI tool:
>
> <pre class="lang-bash">
> sudo rabbitmqctl list_queues
> </pre>
>
> On Windows, omit the sudo:
> <pre class="lang-powershell">
> rabbitmqctl.bat list_queues
> </pre>
Time to move on to [tutorial 2](tutorial-two-steeltoe.html) and build a simple **work queue**.

Просмотреть файл

@ -0,0 +1,29 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
internal const string QueueName = "hello";
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
// Add the rabbit listener
services.AddSingleton<Tut1Receiver>();
services.AddRabbitListeners<Tut1Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,15 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,21 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
namespace Receiver
{
public class Tut1Receiver
{
private readonly ILogger _logger;
public Tut1Receiver(ILogger<Tut1Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = Program.QueueName)]
public void Receive(string input)
{
_logger.LogInformation($"Received: {input}");
}
}
}

Просмотреть файл

@ -0,0 +1,25 @@
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
// The name of the queue that will be created
internal const string QueueName = "hello";
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
services.AddHostedService<Tut1Sender>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-2F9F8745-3AD7-41AD-AF06-AAB67AD8DD50</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,26 @@
using Steeltoe.Messaging.RabbitMQ.Core;
namespace Sender
{
public class Tut1Sender : BackgroundService
{
private readonly ILogger<Tut1Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
public Tut1Sender(ILogger<Tut1Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
await _rabbitTemplate.ConvertAndSendAsync(Program.QueueName, "Hello World!");
_logger.LogInformation("Worker running at: {time}, sent message!", DateTimeOffset.Now);
await Task.Delay(1000, stoppingToken);
}
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{5D219E95-D1D5-445D-AA7A-5EE12D2FC648}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{749C54CB-C3AC-445A-8F64-61A4F7934AB3}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{5D219E95-D1D5-445D-AA7A-5EE12D2FC648}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{5D219E95-D1D5-445D-AA7A-5EE12D2FC648}.Debug|Any CPU.Build.0 = Debug|Any CPU
{5D219E95-D1D5-445D-AA7A-5EE12D2FC648}.Release|Any CPU.ActiveCfg = Release|Any CPU
{5D219E95-D1D5-445D-AA7A-5EE12D2FC648}.Release|Any CPU.Build.0 = Release|Any CPU
{749C54CB-C3AC-445A-8F64-61A4F7934AB3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{749C54CB-C3AC-445A-8F64-61A4F7934AB3}.Debug|Any CPU.Build.0 = Debug|Any CPU
{749C54CB-C3AC-445A-8F64-61A4F7934AB3}.Release|Any CPU.ActiveCfg = Release|Any CPU
{749C54CB-C3AC-445A-8F64-61A4F7934AB3}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {D89895A3-07E9-47A2-987A-96A1B609D0D0}
EndGlobalSection
EndGlobal

Просмотреть файл

@ -0,0 +1,389 @@
# RabbitMQ Tutorial - Work Queues
## Work Queues (using Steeltoe)
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
In the [first tutorial](../Tutorial1/Readme.md) we
wrote programs to send and receive messages from a named queue. In this
tutorial we'll create a _Work Queue_ that will be used to distribute
time-consuming tasks among multiple workers.
The main idea behind Work Queues (aka: _Task Queues_) is to avoid
doing a resource-intensive task immediately and having to wait for
it to complete. Instead we schedule the task to be done later. We encapsulate a
_task_ as a message and send it to a queue. A worker process running
in the background will pop the tasks and eventually execute the
job. When you run many workers the tasks will be shared between them.
This concept is especially useful in web applications where it's
impossible to handle a complex task during a short HTTP request
window.
## Preparation
In the first tutorial we sent a message containing
"Hello World!" as a String. Now we'll be sending strings that stand for complex
tasks. We don't have a real-world task, like images to be resized or
PDF files to be rendered, so let's fake it by just pretending we're
busy - by using the `Thread.Sleep()` function. We'll take the number of dots
in the string as its complexity; every dot will account for one second
of "work". For example, a fake task described by `Hello...`
will take three seconds.
Please see the setup used in [first tutorial](../Tutorial1/Readme.md)
if you have not setup the project. We will follow the same pattern for
all of the rest of the tutorials in this series. As a reminder you should:
- Create a VS2022 solution with an initial `Console` application project which will become the `Receiver`. Add a `Tut2Receiver` class to the project.
- Add a `Worker Service` project to the solution. Name the project `Sender` and rename the `Worker.cs` file to `Tut2Sender.cs`.
- Update the `.csproj` files with the Steeltoe RabbitMQ messaging package reference.
- Update both `Program.cs` files to use the `RabbitMQHost` like we did in the first tutorial.
Here is what the `Program.cs` file for the receiver should look like when you're done:
```csharp
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add the rabbit listener
services.AddSingleton<Tut2Receiver>();
services.AddRabbitListeners<Tut2Receiver>();
})
.Build()
.Run();
}
}
}
```
And here is what the sender `Program.cs` file should look like:
```csharp
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Tut2Sender>();
})
.Build()
.Run();
}
}
}
```
Notice above we did not add the `Queue` to the service container like we did in the first tutorial. Instead we are going to leverage another feature of Steeltoe that enables us to declare `RabbitMQ` entities (i.e. Queues, Exchanges, Bindings, etc) using a declarative approach leveraging .NET attributes. We will see this when we update `Tut2Receiver` below.
## Sender
We will modify the sender to provide a means for identifying
whether it's a longer running task by appending dots to the
message in a very contrived fashion. We will be using the same method
on the `RabbitTemplate` to publish the message `ConvertAndSendAsync()`.
The Steeltoe documentation defines this as, "Convert an object to
a message and send it to a default exchange with a
default routing key."
Here is what the `Tut2Sender` looks like:
```csharp
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut2Sender : BackgroundService
{
private const string QueueName = "hello";
private readonly ILogger<Tut2Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int dots = 0;
private int count = 0;
public Tut2Sender(ILogger<Tut2Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
var message = CreateMessage();
await _rabbitTemplate.ConvertAndSendAsync(QueueName, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
private string CreateMessage()
{
StringBuilder builder = new StringBuilder("Hello");
if (++dots == 4)
{
dots = 1;
}
for (int i = 0; i < dots; i++)
{
builder.Append('.');
}
builder.Append(++count);
return builder.ToString();
}
}
}
```
## Receiver
Our receiver, `Tut2Receiver`, simulates an arbitrary length for
a fake task in the `DoWork()` method where the number of dots
translates into the number of seconds the work will take.
Again, we leverage a `RabbitListener` on a queue named `hello` just like in the first tutorial.
Here is what the code for the `Tut2Receiver` looks like:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using System.Diagnostics;
namespace Receiver
{
[DeclareQueue(Name = "hello")]
internal class Tut2Receiver
{
private readonly ILogger _logger;
public Tut2Receiver(ILogger<Tut2Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@hello}")]
public void Receive(string input)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} took: {time}");
}
private void DoWork(string input)
{
foreach(var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}
```
You should notice a couple new changes in the receiver that you did not see in the first tutorial. First notice the attribute on the `Tut2Receiver` class:
```csharp
[DeclareQueue(Name = "hello")]
```
The above is the declarative way in Steeltoe to add a queue to the service container. In the first tutorial we used the `AddQueue()` method in `Program.cs`; in this tutorial we have switched to using the attribute mechanism instead.
The second change you should see is in how we reference the queue in the `RabbitListener` attribute:
```csharp
[RabbitListener(Queue = "#{@hello}")]
```
This syntax uses a powerful Steeltoe feature that leverages a built in `expression language` that is executed when the listener is created. To use the language, you enclose the `expression` inside a `#{...}` as shown above. In this case the expression is `@hello`. The `@` symbol is part of the language,; it is used to specify a reference to service from the service container is desired and the name of the service in the container follows that `@` symbol. In this case, the service name `hello` is used to reference the `Queue` that was added to the service container using the `DeclareQueue` attribute we mentioned above. This is how the `RabbitListener` ties the `Receive()` method to the `hello` queue.
## Putting it all together
Compile both projects using `dotnet build`.
```bash
cd tutorials\tutorial2
dotnet build
```
Run multiple `Receivers` in different command windows and then start up the sender:
```bash
# receiver1
cd receiver
dotnet run
# receiver2
cd receiver
dotnet run
# sender
cd sender
dotnet run
```
Notice how the work that is produced by the sender is distributed across both receivers.
## Message acknowledgment
Doing a task can take a few seconds, you may wonder what happens if a consumer starts a long task and it terminates before it completes. By default once RabbitMQ delivers a message to the consumer, it immediately marks it for deletion. In this case, if you terminate a worker, the message it was just processing is lost. The messages that were dispatched to this particular worker but were not yet handled are also lost.
But we don't want to lose any tasks. If a worker dies, we'd like the task to be delivered to another worker.
In order to make sure a message is never lost, RabbitMQ supports [message _acknowledgments_](https://www.rabbitmq.com/confirms.html). An acknowledgement is sent back by the consumer to tell RabbitMQ that a particular message has been received, processed and that RabbitMQ is free to delete it.
If a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.
By default Steeltoe takes a conservative approach to [message acknowledgement](https://www.rabbitmq.com/confirms.html). If the listener throws an exception the underlying `Rabbit Container` created by Steeltoe (note: we talked about it the first tutorial) calls:
```csharp
channel.BasicReject(deliveryTag, requeue)
```
Requeue is true by default. This is the typical behavior you want as you don't want to lose any tasks.
But, there are sometimes you want the message to be dropped (i.e. not requeued). You have two ways to control this in Steeltoe. You can explicitly configure the `Container Factory` we mentioned in the first tutorial to default to false for `requeue` when it creates `Rabbit Containers`. Or, the other option is in the `RabbitListener` code you write, you throw a `RabbitRejectAndDoNotRequeueException` instead of some other exception. In this case Steeltoe will not requeue the message and instead just acknowledge it.
Acknowledgements must be sent on the same channel the delivery
was received on. Attempts to acknowledge using a different channel
will result in a channel-level protocol exception. See the [doc guide on confirmations](https://www.rabbitmq.com/confirms.html) to learn more.
Steeltoe generally takes care of this for you, but when used in combination with code
that uses RabbitMQ .NET client directly, this is something to keep in mind.
> #### Forgotten acknowledgments
>
> It's a common programming mistake to miss the `BasicAck` when using the .NET client directly.
> Its an easy error, and the consequences can be serious. Messages will be redelivered
> when your client quits (which may look like random redelivery), but
> RabbitMQ will eat more and more memory as it won't be able to release
> any un-acked messages.
>
> Steeltoe helps to avoid this mistake through its default configuration and managing the acknowledgement for the developer in the `Rabbit Container`.
>
## Message durability
In the previous section we discussed how to make sure that even if the consumer dies, the
task isn't lost. We learned that by default, Steeltoe enables and manages message acknowledgments for the developer.
But our tasks will still be lost if RabbitMQ server stops.
When RabbitMQ quits or crashes it will forget the queues and messages
unless you tell it not to. Two things are required to make sure that
messages aren't lost: we need to mark both the queue and messages as
durable.
Messages are persistent by default with Steeltoe. Note the queue
the message will end up in needs to be durable as well, otherwise
the message will not survive a broker restart as a non-durable queue does not
itself survive a restart. With Steeltoe you can specify the durability of queues using the `Durable` property
on the `DeclareQueue` attribute.
If you want to have more control over the message persistence or over any other aspects of outbound
messages in Steeltoe, you can use `RabbitTemplate#ConvertAndSend(...)` methods
that accept a `IMessagePostProcessor` parameter. `IMessagePostProcessor`
provides a callback before the message is actually sent, so this
is a good place to modify the message payload or any headers that will be sent.
> #### Note on message persistence
>
> Marking messages as persistent doesn't fully guarantee that a message
> won't be lost. Although it tells RabbitMQ to save the message to disk,
> there is still a short time window when RabbitMQ has accepted a message and
> hasn't saved it yet. Also, RabbitMQ doesn't do `fsync(2)` for every
> message -- it may be just saved to cache and not really written to the
> disk. The persistence guarantees aren't strong, but it's more than enough
> for our simple task queue. If you need a stronger guarantee then you can use
> [publisher confirms](https://www.rabbitmq.com/confirms.html).
### Fair dispatch vs Round-robin dispatching
By default, RabbitMQ will send each message to the next consumer,
in sequence. On average every consumer will get the same number of
messages. This way of distributing messages is called round-robin.
With this default RabbitMQ mode, dispatching doesn't necessarily work exactly as we want.
For example in a situation with two workers, when all
odd messages are heavy and even messages are light, one worker will be
constantly busy and the other one will do hardly any work. Well,
RabbitMQ doesn't know anything about that and will still dispatch
messages evenly.
This happens because RabbitMQ just dispatches a message when the message
enters the queue. It doesn't look at the number of unacknowledged
messages for a consumer. It just blindly dispatches every n-th message
to the n-th consumer.
One solution that is commonly recommended is to use a RabbitMQ feature called `prefetchCount` and to set the count to 1.
This tells RabbitMQ not to give more than one message to a worker at a time.
Or, in other words, don't dispatch a new message to a worker until it has processed and acknowledged the previous one.
Instead, it will dispatch any new message to the next worker that is not still busy.
<div class="diagram">
<img src="../img/tutorials/prefetch-count.png" height="110" alt="Producer -> Queue -> Consuming: RabbitMQ dispatching messages." />
</div>
However in most of the cases using a `prefetchCount` equal to 1 would be too conservative and severely
limit consumer throughput.
Instead Steeltoe defaults the `prefetchCount` to 250. This tells RabbitMQ not to give more than 250 messages to a worker
at a time. Or, in other words, don't dispatch a new message to a worker while the number of un-acked messages is 250. This setting improves throughput while also enabling a `Fair Dispatching` of messages.
> #### Note about queue size
>
> If all the workers are busy, your queue can fill up. You will want to keep an
> eye on that, and maybe add more workers, or have some other strategy.
By using Steeltoe Messaging you get reasonable values configured for
message acknowledgments and fair dispatching. The default durability
for queues and persistence for messages provided by Steeltoe
allow the messages to survive even if RabbitMQ is restarted.
Now we can move on to [tutorial 3](tutorial-three-steeltoe.html) and learn how to deliver the same message to many consumers.

Просмотреть файл

@ -0,0 +1,23 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add the rabbit listener
services.AddSingleton<Tut2Receiver>();
services.AddRabbitListeners<Tut2Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,15 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,40 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using System.Diagnostics;
namespace Receiver
{
[DeclareQueue(Name = "hello")]
internal class Tut2Receiver
{
private readonly ILogger _logger;
public Tut2Receiver(ILogger<Tut2Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@hello}")]
public void Receive(string input)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} took: {time}");
}
private void DoWork(string input)
{
foreach(var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}

Просмотреть файл

@ -0,0 +1,18 @@
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Tut2Sender>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-29CDA118-FCEF-4E21-876D-33FB66A1B06E</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,49 @@
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut2Sender : BackgroundService
{
private const string QueueName = "hello";
private readonly ILogger<Tut2Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int dots = 0;
private int count = 0;
public Tut2Sender(ILogger<Tut2Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
var message = CreateMessage();
await _rabbitTemplate.ConvertAndSendAsync(QueueName, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
private string CreateMessage()
{
StringBuilder builder = new StringBuilder("Hello");
if (++dots == 4)
{
dots = 1;
}
for (int i = 0; i < dots; i++)
{
builder.Append('.');
}
builder.Append(++count);
return builder.ToString();
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{9B6765B9-3B76-4271-ABD1-6D3A56351374}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{EA4AA576-CF3E-4F2B-8BE1-22972156FBA5}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{9B6765B9-3B76-4271-ABD1-6D3A56351374}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{9B6765B9-3B76-4271-ABD1-6D3A56351374}.Debug|Any CPU.Build.0 = Debug|Any CPU
{9B6765B9-3B76-4271-ABD1-6D3A56351374}.Release|Any CPU.ActiveCfg = Release|Any CPU
{9B6765B9-3B76-4271-ABD1-6D3A56351374}.Release|Any CPU.Build.0 = Release|Any CPU
{EA4AA576-CF3E-4F2B-8BE1-22972156FBA5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{EA4AA576-CF3E-4F2B-8BE1-22972156FBA5}.Debug|Any CPU.Build.0 = Debug|Any CPU
{EA4AA576-CF3E-4F2B-8BE1-22972156FBA5}.Release|Any CPU.ActiveCfg = Release|Any CPU
{EA4AA576-CF3E-4F2B-8BE1-22972156FBA5}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {4A7D7FF9-4990-467F-A48A-6E9CB73FA805}
EndGlobalSection
EndGlobal

Просмотреть файл

@ -0,0 +1,433 @@
# RabbitMQ tutorial - Publish/Subscribe
## Publish/Subscribe (using Steeltoe)
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
In the [first tutorial](../Tutorial1/Readme.md) we showed how
to use Visual Studio to create a solution with two projects
with the Steeltoe RabbitMQ Messaging dependency and to create simple
applications that send and receive string hello messages.
In the [previous tutorial](../Tutorial2/Readme.md) we created
a sender and receiver and a work queue with two consumers.
We also used Steeltoe attributes to declare the queue.
The assumption behind a work queue is that each task is delivered to exactly one worker.
In this part we'll implement a fanout pattern to deliver
a message to multiple consumers. This pattern is also known as "publish/subscribe"
and is implemented by configuring a number of RabbitMQ entities using Steeltoe attributes.
Essentially, published messages are going to be broadcast to all the receivers.
Exchanges
---------
In previous parts of the tutorial we sent and received messages to and
from a queue. Now it's time to introduce the full messaging model in
RabbitMQ.
Let's quickly go over what we covered in the previous tutorials:
* A _producer_ is a user application that sends messages.
* A _queue_ is a buffer that stores messages.
* A _consumer_ is a user application that receives messages.
The core idea in the messaging model in RabbitMQ is that the producer
never sends any messages directly to a queue. Actually, quite often
the producer doesn't even know if a message will be delivered to any
queue at all.
Instead, the producer can only send messages to an _exchange_. An
exchange is a very simple thing. On one side it receives messages from
producers and the other side it pushes them to queues. The exchange
must know exactly what to do with a message it receives. Should it be
appended to a particular queue? Should it be appended to many queues?
Or should it get discarded. The rules for that are defined by the
_exchange type_.
<div class="diagram">
<img src="../img/tutorials/exchanges.png" height="110" alt="An exchange: The producer can only send messages to an exchange. One side of the exchange receives messages from producers and the other side pushes them to queues."/>
</div>
There are a few exchange types available: `direct`, `topic`, `headers`
and `fanout`. We'll focus on the last one -- the fanout. Let's setup
our Receiver with an exchange of this type, and call it `tut.fanout`:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.fanout", Type = ExchangeType.FANOUT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue1", ExchangeName = "tut.fanout", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue2", ExchangeName = "tut.fanout", QueueName = "#{@queue2}")]
internal class Tut3Receiver
{
private readonly ILogger _logger;
public Tut3Receiver(ILogger<Tut3Receiver> logger)
{
_logger = logger;
}
....
}
}
```
We follow the same approach as in the previous tutorial and use attributes to declare our RabbitMQ entities.
We declare the `FanoutExchange` using the `DeclareExchange` attribute. We also
define four additional RabbitMQ entities, two `AnonymousQueue`s (non-durable, exclusive, auto-delete queues
in AMQP terms) using the `DeclareAnonymousQueue`and two bindings (`DeclareQueueBinding`) to bind those queues to the exchange.
Notice how we tie together these entities. First, the name of the exchange is `tut.fanout` and the two anonymous queues are named `queue1` and `queue2`.
Next, the bindings reference both the exchange name (e.g. `ExchangeName = "tut.fanout"`) and the queue name (e.g. `QueueName = "#{@queue2}"`). Notice we use
the `expression language` we mentioned in the previous tutorial to in the queue name reference.
The fanout exchange is very simple. As you can probably guess from the
name, it just broadcasts all the messages it receives to all the
queues it knows. And that's exactly what we need for fanning out our
messages.
> #### Listing exchanges
>
> To list the exchanges on the server you can run the ever useful `rabbitmqctl`:
>
> ```bash
> sudo rabbitmqctl list_exchanges
> ```
>
> In this list there will be some `amq.*` exchanges and the default (unnamed)
> exchange. These are created by default, but it is unlikely you'll need to
> use them at the moment.
> #### Nameless exchange
>
> In previous parts of the tutorial we knew nothing about exchanges,
> but still were able to send messages to queues. That was possible
> because we were using a default exchange, which we identify by the empty string (`""`).
>
> Recall how we published a message before:
>
> ```csharp
> template.ConvertAndSend(QueueName, message)
>```
>
> The first parameter is the routing key and the `RabbitTemplate`
> sends messages by default to the default exchange. Each queue is automatically
> bound to the default exchange with the name of queue as the binding key.
> This is why we can use the name of the queue as the routing key to make
> sure the message ends up in the queue.
Now, we can publish to our named exchange instead:
```csharp
namespace Sender
{
public class Tut3Sender : BackgroundService
{
internal const string FanoutExchangeName = "tut.fanout";
private readonly RabbitTemplate _rabbitTemplate;
public Tut3Sender(ILogger<Tut3Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
// ....
await _rabbitTemplate.ConvertAndSendAsync(FanoutExchangeName, string.Empty, message);
}
}
}
}
```
From now on the `fanout` exchange will append messages to our queues.
Temporary queues
----------------
As you may remember previously we were using queues that had
specific names (remember `hello`). Being able to name
a queue was crucial for us -- we needed to point the workers to the
same queue. Giving a queue a name is important when you
want to share the queue between producers and consumers.
But that's not the case for our fanout example. We want to hear about
all messages, not just a subset of them. We're
also interested only in currently flowing messages, not in the old
ones. To solve that we need two things.
Firstly, whenever we connect to the broker, we need a fresh, empty queue.
To do this, we could create a queue with a random name, or --
even better -- let the server choose a random queue name for us.
Secondly, once we disconnect the consumer, the queue should be
automatically deleted. To do this with Steeltoe Messaging,
we defined an _AnonymousQueue_, using the `DeclareAnonymousQueue` attribute
which creates a non-durable, exclusive, auto-delete queue with a generated name:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.fanout", Type = ExchangeType.FANOUT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue1", ExchangeName = "tut.fanout", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue2", ExchangeName = "tut.fanout", QueueName = "#{@queue2}")]
internal class Tut3Receiver
{
private readonly ILogger _logger;
public Tut3Receiver(ILogger<Tut3Receiver> logger)
{
_logger = logger;
}
....
}
}
```
At this point, our queues have random queue names. For example,
it may look something like `spring.gen-1Rx9HOqvTAaHeeZrQWu8Pg`.
Bindings
--------
<div class="diagram">
<img src="../img/tutorials/bindings.png" height="90" alt="The exchange sends messages to a queue. The relationship between the exchange and a queue is called a binding." />
</div>
We've already created a fanout exchange and a queue. Now we need to
tell the exchange to send messages to our queue. That relationship
between exchange and a queue is called a _binding_. Below you can see that we have two bindings declared using the `DeclareQueueBinding` attribute , one for each
`AnonymousQueue`.
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.fanout", Type = ExchangeType.FANOUT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue1", ExchangeName = "tut.fanout", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue2", ExchangeName = "tut.fanout", QueueName = "#{@queue2}")]
internal class Tut3Receiver
{
private readonly ILogger _logger;
public Tut3Receiver(ILogger<Tut3Receiver> logger)
{
_logger = logger;
}
....
}
}
```
> #### Listing bindings
>
> You can list existing bindings using, you guessed it,
> ```bash
> rabbitmqctl list_bindings
> ```
Putting it all together
-----------------------
<div>
<img src="../img/tutorials/python-three-overall.png"/>
</div>
The producer program, which emits messages, doesn't look much
different from the previous tutorial. The most important change is that
we now want to publish messages to our `fanout` exchange instead of the
nameless one. We need to supply a `routingKey` when sending, but its
value is ignored for `fanout` exchanges.
Here goes the code for `Tut3Sender.cs` program:
```csharp
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut3Sender : BackgroundService
{
private readonly ILogger<Tut3Sender> _logger;
private int dots = 0;
private int count = 0;
internal const string FanoutExchangeName = "tut.fanout";
private readonly RabbitTemplate _rabbitTemplate;
public Tut3Sender(ILogger<Tut3Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
var message = CreateMessage();
await _rabbitTemplate.ConvertAndSendAsync(FanoutExchangeName, string.Empty, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
private string CreateMessage()
{
StringBuilder builder = new StringBuilder("Hello");
if (++dots == 4)
{
dots = 1;
}
for (int i = 0; i < dots; i++)
{
builder.Append('.');
}
builder.Append(++count);
return builder.ToString();
}
}
}
```
As you see, we leverage dependency injection and add the `RabbitTemplate` to the constructors signature.
Note that messages will be lost if no queue is bound to the exchange yet,
but that's okay for us; if no consumer is listening yet we can safely discard the message.
Next the code for `Tut3Receiver.cs`:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.fanout", Type = ExchangeType.FANOUT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue1", ExchangeName = "tut.fanout", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue2", ExchangeName = "tut.fanout", QueueName = "#{@queue2}")]
internal class Tut3Receiver
{
private readonly ILogger _logger;
public Tut3Receiver(ILogger<Tut3Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@queue1}")]
public void Receive1(string input)
{
Receive(input, 1);
}
[RabbitListener(Queue = "#{@queue2}")]
public void Receive2(string input)
{
Receive(input, 2);
}
private void Receive(string input, int receiver)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} from queue: {receiver}, took: {time}");
}
private void DoWork(string input)
{
foreach (var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}
```
Compile as before and we're ready to execute the fanout sender and receiver.
```bash
cd tutorials\tutorial3
dotnet build
```
And of course, to execute the tutorial do the following:
To run the receiver, execute the following commands:
```bash
# receiver
cd receiver
dotnet run
```
Open another shell to run the sender:
```bash
# sender
cd sender
dotnet run
```
Using `rabbitmqctl list_bindings` you can verify that the code actually
creates bindings and queues as we want. With two `ReceiveLogs.java`
programs running you should see something like:
To find out how to listen for a subset of messages, let's move on to
[tutorial 4](../Tutorial4/readme.md)

Просмотреть файл

@ -0,0 +1,24 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add the rabbit listener
services.AddSingleton<Tut3Receiver>();
services.AddRabbitListeners<Tut3Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,15 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,58 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.fanout", Type = ExchangeType.FANOUT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue1", ExchangeName = "tut.fanout", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.fanout.binding.queue2", ExchangeName = "tut.fanout", QueueName = "#{@queue2}")]
internal class Tut3Receiver
{
private readonly ILogger _logger;
public Tut3Receiver(ILogger<Tut3Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@queue1}")]
public void Receive1(string input)
{
Receive(input, 1);
}
[RabbitListener(Queue = "#{@queue2}")]
public void Receive2(string input)
{
Receive(input, 2);
}
private void Receive(string input, int receiver)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} from queue: {receiver}, took: {time}");
}
private void DoWork(string input)
{
foreach (var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}

Просмотреть файл

@ -0,0 +1,17 @@
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Tut3Sender>();
})
.Build().Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-C2D2E18A-2CEB-41A7-B14E-12B4C99EDB13</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,49 @@
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut3Sender : BackgroundService
{
private readonly ILogger<Tut3Sender> _logger;
private int dots = 0;
private int count = 0;
internal const string FanoutExchangeName = "tut.fanout";
private readonly RabbitTemplate _rabbitTemplate;
public Tut3Sender(ILogger<Tut3Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
var message = CreateMessage();
await _rabbitTemplate.ConvertAndSendAsync(FanoutExchangeName, string.Empty, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
private string CreateMessage()
{
StringBuilder builder = new StringBuilder("Hello");
if (++dots == 4)
{
dots = 1;
}
for (int i = 0; i < dots; i++)
{
builder.Append('.');
}
builder.Append(++count);
return builder.ToString();
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{706B9EBD-115F-4A58-B616-AC35C7AA470D}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{1BF76297-BC98-4C89-91E9-A06FEB0AC779}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{706B9EBD-115F-4A58-B616-AC35C7AA470D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{706B9EBD-115F-4A58-B616-AC35C7AA470D}.Debug|Any CPU.Build.0 = Debug|Any CPU
{706B9EBD-115F-4A58-B616-AC35C7AA470D}.Release|Any CPU.ActiveCfg = Release|Any CPU
{706B9EBD-115F-4A58-B616-AC35C7AA470D}.Release|Any CPU.Build.0 = Release|Any CPU
{1BF76297-BC98-4C89-91E9-A06FEB0AC779}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{1BF76297-BC98-4C89-91E9-A06FEB0AC779}.Debug|Any CPU.Build.0 = Debug|Any CPU
{1BF76297-BC98-4C89-91E9-A06FEB0AC779}.Release|Any CPU.ActiveCfg = Release|Any CPU
{1BF76297-BC98-4C89-91E9-A06FEB0AC779}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {C635DCAA-3C1D-4D0C-BB99-A53064918345}
EndGlobalSection
EndGlobal

Просмотреть файл

@ -0,0 +1,305 @@
# RabbitMQ Tutorial - Routing
## Routing (using Steeltoe)
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
In the [previous tutorial](../Tutorial3/readme.md) we built a
simple fanout exchange. We were able to broadcast messages to many
receivers.
In this tutorial we're going to add a feature to it - we're going to
make it possible to subscribe only to a subset of the messages. For
example, we will be able to direct only messages to the
certain colors of interest ("orange", "black", "green"), while still being
able to print all of the messages on the console.
Bindings
--------
In previous examples we were already creating bindings. You may recall
code like this:
```csharp
[DeclareQueueBinding(Name = "tut.fanout.binding.queue1", ExchangeName = "tut.fanout", QueueName = "#{@queue1}")]
```
Remember, a binding is a relationship between an exchange and a queue. This can
be simply read as: the queue is interested in messages from this
exchange.
Bindings can take an extra routing key parameter which we didn't use in the previous tutorial.
We can specify the key using the `RoutingKey` property on the `DeclareQueueBinding` attribute as shown below:
```csharp
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.orange", ExchangeName = "tut.direct", RoutingKey = "orange", QueueName = "#{@queue1}")]
```
The meaning of a routing key depends on the exchange type. The
`fanout` exchanges, which we used previously, simply ignored its
value.
Direct exchange
---------------
Our messaging system from the previous tutorial broadcasts all messages
to all consumers. We want to extend that to allow filtering messages
based on their color type. For example, we may want a program which
writes log messages to the disk to only receive critical errors, and
not waste disk space on warning or info log messages.
Before, we were using a `fanout` exchange, which doesn't give us much
flexibility - it's only capable of mindless broadcasting.
In this tutorial We will use a `direct` exchange instead. The routing algorithm behind
a `direct` exchange is simple - a message goes to the queues whose
binding key exactly matches the routing key of the message.
To illustrate that, consider the following setup:
<div class="diagram">
<img src="../img/tutorials/direct-exchange.png" height="170" alt="Direct Exchange routing" />
</div>
In this setup, we can see the `direct` exchange `X` with two queues bound
to it. The first queue is bound with binding key `orange`, and the second
has two bindings, one with binding key `black` and the other one
with `green`.
In such a setup a message published to the exchange with a routing key
`orange` will be routed to queue `Q1`. Messages with a routing key of `black`
or `green` will go to `Q2`. All other messages will be discarded.
Multiple bindings
-----------------
<div class="diagram">
<img src="../img/tutorials/direct-exchange-multiple.png" height="170" alt="Multiple Bindings" />
</div>
It is perfectly legal to bind multiple queues with the same binding
key. In our example we could add a binding between `X` and `Q1` with
binding key `black`. In that case, the `direct` exchange will behave
like `fanout` and will broadcast the message to all the matching
queues. A message with routing key `black` will be delivered to both
`Q1` and `Q2`.
Here are the `DeclareQueueBinding` attributes that illustrate the above concepts.
```csharp
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.orange", ExchangeName = "tut.direct", RoutingKey = "orange", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.green", ExchangeName = "tut.direct", RoutingKey = "green", QueueName = "#{@queue2}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue2}")]
```
Publishing messages
-------------
We'll use this model for our routing system. Instead of `fanout` we'll
send messages to a `direct` exchange defined using the attribute shown below:
```csharp
[DeclareExchange(Name = "tut.direct", Type = ExchangeType.DIRECT)]
```
We will supply the color as a routing key in the `ConvertAndSendAsync()` method call. That way the receiving program will be able to select
the color it wants to receive (or subscribe to).
Subscribing
-----------
Receiving messages will work just like in the previous tutorial, with
one exception - we're going to create a new binding for each color
we're interested in.
Here's what that looks like in the `Tut4Receiver`:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.direct", Type = ExchangeType.DIRECT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.orange", ExchangeName = "tut.direct", RoutingKey = "orange", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.green", ExchangeName = "tut.direct", RoutingKey = "green", QueueName = "#{@queue2}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue2}")]
internal class Tut4Receiver
{
private readonly ILogger _logger;
public Tut4Receiver(ILogger<Tut4Receiver> logger)
{
_logger = logger;
}
....
}
}
```
Putting it all together
-----------------------
<div class="diagram">
<img src="../img/tutorials/python-four.png" height="170" alt="Final routing: putting it all together." />
</div>
The code for our sender class (`Tut4Sender`) is:
```csharp
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut4Sender : BackgroundService
{
internal const string DirectExchangeName = "tut.direct";
private readonly ILogger<Tut4Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int index = 0;
private int count = 0;
private readonly string[] keys = new string[] { "orange", "black", "green" };
public Tut4Sender(ILogger<Tut4Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
StringBuilder builder = new StringBuilder("Hello to ");
if (++index == 3)
{
index = 0;
}
string key = keys[index];
builder.Append(key).Append(' ');
builder.Append(++count);
var message = builder.ToString();
await _rabbitTemplate.ConvertAndSendAsync(DirectExchangeName, key, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
}
}
```
The code for receiver class (`Tut4Receiver`) is:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.direct", Type = ExchangeType.DIRECT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.orange", ExchangeName = "tut.direct", RoutingKey = "orange", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.green", ExchangeName = "tut.direct", RoutingKey = "green", QueueName = "#{@queue2}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue2}")]
internal class Tut4Receiver
{
private readonly ILogger _logger;
public Tut4Receiver(ILogger<Tut4Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@queue1}")]
public void Receive1(string input)
{
Receive(input, 1);
}
[RabbitListener(Queue = "#{@queue2}")]
public void Receive2(string input)
{
Receive(input, 2);
}
private void Receive(string input, int receiver)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} from queue: {receiver}, took: {time}");
}
private void DoWork(string input)
{
foreach (var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}
```
Compile as usual, see [tutorial one](../Tutorial1/Readme.md)
```bash
cd tutorials\tutorial4
dotnet build
```
To run the receiver, execute the following commands:
```bash
# receiver
cd receiver
dotnet run
```
Open another shell to run the sender:
```bash
# sender
cd sender
dotnet run
```
Move on to [tutorial 5](../Tutorial5/readme.md) to find out how to listen
for messages based on a pattern.

Просмотреть файл

@ -0,0 +1,24 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add the rabbit listener
services.AddSingleton<Tut4Receiver>();
services.AddRabbitListeners<Tut4Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,15 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,59 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = "tut.direct", Type = ExchangeType.DIRECT)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.orange", ExchangeName = "tut.direct", RoutingKey = "orange", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue1.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.green", ExchangeName = "tut.direct", RoutingKey = "green", QueueName = "#{@queue2}")]
[DeclareQueueBinding(Name = "tut.direct.binding.queue2.black", ExchangeName = "tut.direct", RoutingKey = "black", QueueName = "#{@queue2}")]
internal class Tut4Receiver
{
private readonly ILogger _logger;
public Tut4Receiver(ILogger<Tut4Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@queue1}")]
public void Receive1(string input)
{
Receive(input, 1);
}
[RabbitListener(Queue = "#{@queue2}")]
public void Receive2(string input)
{
Receive(input, 2);
}
private void Receive(string input, int receiver)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} from queue: {receiver}, took: {time}");
}
private void DoWork(string input)
{
foreach (var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}

Просмотреть файл

@ -0,0 +1,17 @@
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Tut4Sender>();
})
.Build().Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,15 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-B7957B91-4F0F-4F6A-BF26-7DB15BCBF4D6</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,46 @@
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut4Sender : BackgroundService
{
internal const string DirectExchangeName = "tut.direct";
private readonly ILogger<Tut4Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int index = 0;
private int count = 0;
private readonly string[] keys = new string[] { "orange", "black", "green" };
public Tut4Sender(ILogger<Tut4Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
StringBuilder builder = new StringBuilder("Hello to ");
if (++index == 3)
{
index = 0;
}
string key = keys[index];
builder.Append(key).Append(' ');
builder.Append(++count);
var message = builder.ToString();
await _rabbitTemplate.ConvertAndSendAsync(DirectExchangeName, key, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{C4000E5D-5B38-4462-BC6E-B801D4B15D10}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{75D20A25-6377-4EBA-8F21-15C3E7EE485D}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{C4000E5D-5B38-4462-BC6E-B801D4B15D10}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{C4000E5D-5B38-4462-BC6E-B801D4B15D10}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C4000E5D-5B38-4462-BC6E-B801D4B15D10}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C4000E5D-5B38-4462-BC6E-B801D4B15D10}.Release|Any CPU.Build.0 = Release|Any CPU
{75D20A25-6377-4EBA-8F21-15C3E7EE485D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{75D20A25-6377-4EBA-8F21-15C3E7EE485D}.Debug|Any CPU.Build.0 = Debug|Any CPU
{75D20A25-6377-4EBA-8F21-15C3E7EE485D}.Release|Any CPU.ActiveCfg = Release|Any CPU
{75D20A25-6377-4EBA-8F21-15C3E7EE485D}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {3118EF1E-A208-4C86-AEAC-35AC83BB1224}
EndGlobalSection
EndGlobal

Просмотреть файл

@ -0,0 +1,299 @@
# RabbitMQ Tutorial - Topics
## Topics (using Steeltoe)
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
In the [previous tutorial](../Tutorial4/readme.md) we improved our
messaging flexibility. Instead of using a `fanout` exchange only capable of
dummy broadcasting, we used a `direct` one, and gained a possibility
of selectively receiving the message based on the routing key.
Although using the `direct` exchange improved our system, it still has
limitations - it can't do routing based on multiple criteria.
In our messaging system we might want to subscribe to not only queues
based on the routing key, but also based on the source which produced
the message.
You might know this concept from the
[`syslog`](http://en.wikipedia.org/wiki/Syslog) unix tool, which
routes logs based on both severity (info/warn/crit...) and facility
(auth/cron/kern...). Our example is a little simpler than this.
That example would give us a lot of flexibility - we may want to listen to
just critical errors coming from 'cron' but also all logs from 'kern'.
To implement that flexibility in our logging system we need to learn
about a more complex `topic` exchange.
Topic exchange
--------------
Messages sent to a `topic` exchange can't have an arbitrary
`routing_key` - it must be a list of words, delimited by dots. The
words can be anything, but usually they specify some features
connected to the message. A few valid routing key examples:
"`stock.usd.nyse`", "`nyse.vmw`", "`quick.orange.rabbit`". There can be as
many words in the routing key as you like, up to the limit of 255
bytes.
The routing key associated with a binding must also be in the same form. The logic behind the
`topic` exchange is similar to a `direct` one - a message sent with a
particular routing key will be delivered to all the queues that are
bound with a matching binding key. However there are two important
special cases for routing keys associated with bindings.
* `*` (star) can substitute for exactly one word.
* `#` (hash) can substitute for zero or more words.
It's easiest to explain this in an example:
<div class="diagram">
<img src="../img/tutorials/python-five.png" height="170" alt="Topic Exchange illustration, which is all explained in the following text." title="Topic Exchange Illustration" />
</div>
In this example, we're going to send messages which all describe
animals. The messages will be sent with a routing key that consists of
three words (two dots). The first word in the routing key
will describe speed, second a color and third a species:
"`<speed>.<color>.<species>`".
We created three bindings: Q1 is bound with binding key "`*.orange.*`"
and Q2 with "`*.*.rabbit`" and "`lazy.#`".
These bindings can be summarized as:
* Q1 is interested in all the orange animals.
* Q2 wants to hear everything about rabbits, and everything about lazy
animals.
A message with a routing key set to "`quick.orange.rabbit`"
will be delivered to both queues. Message
"`lazy.orange.elephant`" also will go to both of them. On the other hand
"`quick.orange.fox`" will only go to the first queue, and
"`lazy.brown.fox`" only to the second. "`lazy.pink.rabbit`" will
be delivered to the second queue only once, even though it matches two bindings.
"`quick.brown.fox`" doesn't match any binding so it will be discarded.
What happens if we break our contract and send a message with one or
four words, like "`orange`" or "`quick.orange.new.rabbit`"? Well,
these messages won't match any bindings and will be lost.
On the other hand "`lazy.orange.new.rabbit`", even though it has four
words, will match the last binding and will be delivered to the second
queue.
> #### Topic exchange
>
> Topic exchange is powerful and can behave like other exchanges.
>
> When a queue is bound with "`#`" (hash) binding key - it will receive
> all the messages, regardless of the routing key - like in `fanout` exchange.
>
> When special characters "`*`" (star) and "`#`" (hash) aren't used in bindings,
> the topic exchange will behave just like a `direct` one.
Putting it all together
-----------------------
We're going to use a `topic` exchange in our messaging system. We'll
start off with a working assumption that the routing keys will take
advantage of both wildcards and a hash tag.
The code is almost the same as in the
[previous tutorial](../Tutorial4/readme.md).
First let's configure all the RabbitMQ entities using the Steeltoe attributes:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = Program.TopicExchangeName, Type = ExchangeType.TOPIC)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "binding.queue1.orange", ExchangeName = Program.TopicExchangeName, RoutingKey = "*.orange.*", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "binding.queue1.rabbit", ExchangeName = Program.TopicExchangeName, RoutingKey = "*.*.rabbit", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "binding.queue2.lazy", ExchangeName = Program.TopicExchangeName, RoutingKey = "lazy.#", QueueName = "#{@queue2}")]
internal class Tut5Receiver
{
private readonly ILogger _logger;
public Tut5Receiver(ILogger<Tut5Receiver> logger)
{
_logger = logger;
}
......
}
}
```
The `Tut5Receiver` again uses the `RabbitListener` attribute to receive messages from the respective
topics.
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = Program.TopicExchangeName, Type = ExchangeType.TOPIC)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "binding.queue1.orange", ExchangeName = Program.TopicExchangeName, RoutingKey = "*.orange.*", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "binding.queue1.rabbit", ExchangeName = Program.TopicExchangeName, RoutingKey = "*.*.rabbit", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "binding.queue2.lazy", ExchangeName = Program.TopicExchangeName, RoutingKey = "lazy.#", QueueName = "#{@queue2}")]
internal class Tut5Receiver
{
private readonly ILogger _logger;
public Tut5Receiver(ILogger<Tut5Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@queue1}")]
public void Receive1(string input)
{
Receive(input, 1);
}
[RabbitListener(Queue = "#{@queue2}")]
public void Receive2(string input)
{
Receive(input, 2);
}
private void Receive(string input, int receiver)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} from queue: {receiver}, took: {time}");
}
private void DoWork(string input)
{
foreach (var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}
```
The code for `Tut5Sender`:
```csharp
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut5Sender : BackgroundService
{
internal const string TopicExchangeName = "tut.topic";
private readonly ILogger<Tut5Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int index = 0;
private int count = 0;
private readonly string[] keys = new string[] {
"quick.orange.rabbit",
"lazy.orange.elephant",
"quick.orange.fox",
"lazy.brown.fox",
"lazy.pink.rabbit",
"quick.brown.fox"};
public Tut5Sender(ILogger<Tut5Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
StringBuilder builder = new StringBuilder("Hello to ");
if (++index == keys.Length)
{
index = 0;
}
string key = keys[index];
builder.Append(key).Append(' ');
builder.Append(++count);
var message = builder.ToString();
await _rabbitTemplate.ConvertAndSendAsync(TopicExchangeName, key, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
}
}
```
Compile as usual, see [tutorial one](../Tutorial1/readme.md)
```bash
cd tutorials\tutorial5
dotnet build
```
To run the receiver, execute the following commands:
```bash
# receiver
cd receiver
dotnet run
```
Open another shell to run the sender:
```bash
# sender
cd sender
dotnet run
```
Have fun playing with these programs. Note that the code doesn't make
any assumption about the routing or binding keys, you may want to play
with more than two routing key parameters.
Next, find out how to do a round trip message as a remote procedure call (RPC)
in [tutorial 6](../Tutorial6/readme.md)

Просмотреть файл

@ -0,0 +1,25 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
internal const string TopicExchangeName = "tut.topic";
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add the rabbit listener
services.AddSingleton<Tut5Receiver>();
services.AddRabbitListeners<Tut5Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,59 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
using System.Diagnostics;
namespace Receiver
{
[DeclareExchange(Name = Program.TopicExchangeName, Type = ExchangeType.TOPIC)]
[DeclareAnonymousQueue("queue1")]
[DeclareAnonymousQueue("queue2")]
[DeclareQueueBinding(Name = "binding.queue1.orange", ExchangeName = Program.TopicExchangeName, RoutingKey = "*.orange.*", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "binding.queue1.rabbit", ExchangeName = Program.TopicExchangeName, RoutingKey = "*.*.rabbit", QueueName = "#{@queue1}")]
[DeclareQueueBinding(Name = "binding.queue2.lazy", ExchangeName = Program.TopicExchangeName, RoutingKey = "lazy.#", QueueName = "#{@queue2}")]
internal class Tut5Receiver
{
private readonly ILogger _logger;
public Tut5Receiver(ILogger<Tut5Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "#{@queue1}")]
public void Receive1(string input)
{
Receive(input, 1);
}
[RabbitListener(Queue = "#{@queue2}")]
public void Receive2(string input)
{
Receive(input, 2);
}
private void Receive(string input, int receiver)
{
var watch = new Stopwatch();
watch.Start();
DoWork(input);
watch.Stop();
var time = watch.Elapsed;
_logger.LogInformation($"Received: {input} from queue: {receiver}, took: {time}");
}
private void DoWork(string input)
{
foreach (var ch in input)
{
if (ch == '.')
Thread.Sleep(1000);
}
}
}
}

Просмотреть файл

@ -0,0 +1,17 @@
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Tut5Sender>();
})
.Build().Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-98F382B2-A4E4-4F1F-A675-D9A2E3D611BC</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,52 @@
using Steeltoe.Messaging.RabbitMQ.Core;
using System.Text;
namespace Sender
{
public class Tut5Sender : BackgroundService
{
internal const string TopicExchangeName = "tut.topic";
private readonly ILogger<Tut5Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int index = 0;
private int count = 0;
private readonly string[] keys = new string[] {
"quick.orange.rabbit",
"lazy.orange.elephant",
"quick.orange.fox",
"lazy.brown.fox",
"lazy.pink.rabbit",
"quick.brown.fox"};
public Tut5Sender(ILogger<Tut5Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
StringBuilder builder = new StringBuilder("Hello to ");
if (++index == keys.Length)
{
index = 0;
}
string key = keys[index];
builder.Append(key).Append(' ');
builder.Append(++count);
var message = builder.ToString();
await _rabbitTemplate.ConvertAndSendAsync(TopicExchangeName, key, message);
_logger.LogInformation($"Sent '" + message + "'");
await Task.Delay(1000, stoppingToken);
}
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{334ACB3A-F425-426F-929E-2B92A6EA5F7D}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{3A1ADF7E-4E0A-4456-94EC-FCD1F897E1EC}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{334ACB3A-F425-426F-929E-2B92A6EA5F7D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{334ACB3A-F425-426F-929E-2B92A6EA5F7D}.Debug|Any CPU.Build.0 = Debug|Any CPU
{334ACB3A-F425-426F-929E-2B92A6EA5F7D}.Release|Any CPU.ActiveCfg = Release|Any CPU
{334ACB3A-F425-426F-929E-2B92A6EA5F7D}.Release|Any CPU.Build.0 = Release|Any CPU
{3A1ADF7E-4E0A-4456-94EC-FCD1F897E1EC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{3A1ADF7E-4E0A-4456-94EC-FCD1F897E1EC}.Debug|Any CPU.Build.0 = Debug|Any CPU
{3A1ADF7E-4E0A-4456-94EC-FCD1F897E1EC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{3A1ADF7E-4E0A-4456-94EC-FCD1F897E1EC}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {5C1E0445-B538-4F88-B3A9-394B9D75ABD6}
EndGlobalSection
EndGlobal

Просмотреть файл

@ -0,0 +1,302 @@
# RabbitMQ Tutorial - Remote procedure call (RPC)
## Remote procedure call (using Steeltoe)
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
In the [second tutorial](../Tutorial2/readme.md) we learned how to
use _Work Queues_ to distribute time-consuming tasks among multiple
workers.
But what if we need to run a function on a remote computer and wait for
the result? Well, that's a different story. This pattern is commonly
known as _Remote Procedure Call_ or _RPC_.
In this tutorial we're going to use RabbitMQ to build an RPC system: a
client and a scalable RPC server. As we don't have any time-consuming
tasks that are worth distributing, we're going to create a dummy RPC
service that returns Fibonacci numbers.
## Client interface
Normally when we talk about RPC's, we talk in terms of an RPC "Client" and "Server".
In the context of sending and receiving, our Sender will become the RPC "Client" and our Receiver will be our RPC "Server".
When the sender calls the server we will get back the fibonacci of the argument we call with. Here is how the sender will use
the `RabbitTemplate` to invoke the server.
```csharp
int result = await _rabbitTemplate.ConvertSendAndReceiveAsync<int>(RPCExchangeName, "rpc", start++);
_logger.LogInformation($"Got result: {result}");
```
> #### A note on RPC
>
> Although RPC is a pretty common pattern in computing, it's often criticised.
> The problems arise when a programmer is not aware
> whether a function call is local or if it's a slow RPC. Confusions
> like that result in an unpredictable system and adds unnecessary
> complexity to debug. Instead of simplifying software, misused RPC
> can result in unmaintainable spaghetti code.
>
> Bearing that in mind, consider the following advice:
>
> * Make sure it's obvious which function call is local and which is remote.
> * Document your system. Make the dependencies between components clear.
> * Handle error cases. How should the client react when the RPC server is
> down for a long time?
>
> When in doubt avoid RPC. If you can, you should use an asynchronous
> pipeline - instead of RPC-like blocking, results are asynchronously
> pushed to a next computation stage.
## Callback queue
In general doing RPC over RabbitMQ is easy. A client sends a request
message and a server replies with a response message. In order to
receive a response we need to send a 'callback' queue address with the
request. Steeltoe's `RabbitTemplate` handles the callback queue for
us when we use the above `ConvertSendAndReceiveAsync()` method. There is
no need to do any other setup when using the `RabbitTemplate`.
For a thorough explanation please see [Request/Reply Message](https://docs.steeltoe.io/api/v3/messaging/rabbitmq-reference.html#request-and-reply-messaging).
> #### Message properties
>
> The AMQP 0-9-1 protocol predefines a set of 14 properties that go with
> a message. Most of the properties are rarely used, with the exception of
> the following:
>
> * `deliveryMode`: Marks a message as persistent (with a value of `2`)
> or transient (any other value). You may remember this property
> from [the second tutorial](../Tutorial2/readme.md).
> * `contentType`: Used to describe the mime-type of the encoding.
> For example for the often used JSON encoding it is a good practice
> to set this property to: `application/json`.
> * `replyTo`: Commonly used to name a callback queue.
> * `correlationId`: Useful to correlate RPC responses with requests.
## Correlation Id
Steeltoe allows you to focus on the message style you're working
with and hide the details of message plumbing required to support
this style. For example, typically the native client would
create a callback queue for every RPC request. That's pretty
inefficient so an alternative is to create a single callback
queue per client.
That raises a new issue, having received a response in that queue it's
not clear to which request the response belongs. That's when the
`correlationId` property is used. Steeltoe automatically sets
a unique value for every request. In addition it handles the details
of matching the response with the correct correlationID.
One reason that Steeltoe makes RPC style easier over RabbitMQ is that sometimes
you may want to ignore unknown messages in the callback
queue, rather than failing with an error. It's due to a possibility of
a race condition on the server side. Although unlikely, it is possible
that the RPC server will die just after sending us the answer, but
before sending an acknowledgment message for the request. If that
happens, the restarted RPC server will process the request again.
Steeltoe handles the duplicate responses gracefully,
and the RPC should ideally be idempotent.
### Summary
<div class="diagram">
<img src="../img/tutorials/python-six.png" height="200" alt="Summary illustration, which is described in the following bullet points." />
</div>
Our RPC will work like this:
* We will setup a new `DirectExchange`
* The client will leverage the `ConvertSendAndReceive` method, passing the exchange
name, the routingKey, and the message.
* The request is sent to an RPC queue `tut.rpc`.
* The RPC worker (i.e. Server) is waiting for requests on that queue.
When a request appears, it performs the task and returns a message with the
result back to the client, using the queue from the `replyTo` field.
* The client waits for data on the callback queue. When a message
appears, it checks the `correlationId` property. If it matches
the value from the request it returns the response to the
application. Again, this is done automagically via the Steeltoe `RabbitTemplate`.
Putting it all together
-----------------------
The Fibonacci task is a `RabbitListener` and is defined as:
```csharp
[RabbitListener(Queue = "tut.rpc.requests")]
// [SendTo("tut.rpc.replies")] used when the client doesn't set replyTo.
public int Fibonacci(int n)
{
_logger.LogInformation($"Received request for {n}");
var result = Fib(n);
_logger.LogInformation($"Returning {result}");
return result;
}
private int Fib(int n)
{
return n == 0 ? 0 : n == 1 ? 1 : (Fib(n - 1) + Fib(n - 2));
}
```
We declare our Fibonacci function. It assumes only valid positive integer input.
(Don't expect this one to work for big numbers,
and it's probably the slowest recursive implementation possible).
The code to configure the RabbitMQ entities looks like this:
```csharp
[DeclareQueue(Name = "tut.rpc.requests")]
[DeclareExchange(Name = Program.RPCExchangeName, Type = ExchangeType.DIRECT)]
[DeclareQueueBinding(Name ="binding.rpc.queue.exchange", QueueName = "tut.rpc.requests", ExchangeName = Program.RPCExchangeName, RoutingKey = "rpc")]
```
The server code is rather straightforward:
* As usual we start annotating our receiver method with a `RabbitListener`
and defining the RabbitMQ entities using the [Declare****()] attributes
* Our Fibonacci method calls Fib() with the payload parameter and returns
the result
The code for our RPC server:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
namespace Receiver
{
[DeclareQueue(Name = "tut.rpc.requests")]
[DeclareExchange(Name = Program.RPCExchangeName, Type = ExchangeType.DIRECT)]
[DeclareQueueBinding(Name ="binding.rpc.queue.exchange", QueueName = "tut.rpc.requests", ExchangeName = Program.RPCExchangeName, RoutingKey = "rpc")]
internal class Tut6Receiver
{
private readonly ILogger _logger;
public Tut6Receiver(ILogger<Tut6Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "tut.rpc.requests")]
// [SendTo("tut.rpc.replies")] used when the client doesn't set replyTo.
public int Fibonacci(int n)
{
_logger.LogInformation($"Received request for {n}");
var result = Fib(n);
_logger.LogInformation($"Returning {result}");
return result;
}
private int Fib(int n)
{
return n == 0 ? 0 : n == 1 ? 1 : (Fib(n - 1) + Fib(n - 2));
}
}
}
```
The client code is as easy as the server:
* We inject the `RabbitTemplate` service
* We invoke `template.ConvertSendAndReceiveAsync()` with the parameters
exchange name, routing key and message.
* We print the result
```csharp
using Steeltoe.Messaging.RabbitMQ.Core;
namespace Sender
{
public class Tut6Sender : BackgroundService
{
internal const string RPCExchangeName = "tut.rpc";
private readonly ILogger<Tut6Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int start = 0;
public Tut6Sender(ILogger<Tut6Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
_logger.LogInformation($"Requesting Fib({start})");
int result = await _rabbitTemplate.ConvertSendAndReceiveAsync<int>(RPCExchangeName, "rpc", start++);
_logger.LogInformation($"Got result: {result}");
await Task.Delay(1000, stoppingToken);
}
}
}
}
```
Compile as usual, see [tutorial one](../Tutorial1/Readme.md)
```bash
cd tutorials\tutorial6
dotnet build
```
To run the server, execute the following commands:
```bash
# server
cd receiver
dotnet run
```
To request a fibonacci number run the client:
```bash
# client
cd sender
dotnet run
```
The design presented here is not the only possible implementation of a RPC
service, but it has some important advantages:
* If the RPC server is too slow, you can scale up by just running
another one. Try running a second `RPC Server` in a new console.
* On the client side, the RPC requires sending and
receiving only one message with one method. No synchronous calls
like `queueDeclare` are required. As a result the RPC client needs
only one network round trip for a single RPC request.
Our code is still pretty simplistic and doesn't try to solve more
complex (but important) problems, like:
* How should the client react if there are no servers running?
* Should a client have some kind of timeout for the RPC?
* If the server malfunctions and raises an exception, should it be
forwarded to the client?
* Protecting against invalid incoming messages
(eg checking bounds, type) before processing.
Next, find out how to use publisher confirms
in [tutorial 7](../Tutorial7/readme.md)

Просмотреть файл

@ -0,0 +1,24 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
internal const string RPCExchangeName = "tut.rpc";
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add the rabbit listener
services.AddSingleton<Tut6Receiver>();
services.AddRabbitListeners<Tut6Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,34 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using Steeltoe.Messaging.RabbitMQ.Config;
namespace Receiver
{
[DeclareQueue(Name = "tut.rpc.requests")]
[DeclareExchange(Name = Program.RPCExchangeName, Type = ExchangeType.DIRECT)]
[DeclareQueueBinding(Name ="binding.rpc.queue.exchange", QueueName = "tut.rpc.requests", ExchangeName = Program.RPCExchangeName, RoutingKey = "rpc")]
internal class Tut6Receiver
{
private readonly ILogger _logger;
public Tut6Receiver(ILogger<Tut6Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = "tut.rpc.requests")]
// [SendTo("tut.rpc.replies")] used when the client doesn't set replyTo.
public int Fibonacci(int n)
{
_logger.LogInformation($"Received request for {n}");
var result = Fib(n);
_logger.LogInformation($"Returning {result}");
return result;
}
private int Fib(int n)
{
return n == 0 ? 0 : n == 1 ? 1 : (Fib(n - 1) + Fib(n - 2));
}
}
}

Просмотреть файл

@ -0,0 +1,16 @@
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
services.AddHostedService<Tut6Sender>();
})
.Build().Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-CD180ADB-D917-4FB7-BB6C-8891BC414B49</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,30 @@
using Steeltoe.Messaging.RabbitMQ.Core;
namespace Sender
{
public class Tut6Sender : BackgroundService
{
internal const string RPCExchangeName = "tut.rpc";
private readonly ILogger<Tut6Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int start = 0;
public Tut6Sender(ILogger<Tut6Sender> logger, RabbitTemplate rabbitTemplate)
{
_logger = logger;
_rabbitTemplate = rabbitTemplate;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
_logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
_logger.LogInformation($"Requesting Fib({start})");
int result = await _rabbitTemplate.ConvertSendAndReceiveAsync<int>(RPCExchangeName, "rpc", start++);
_logger.LogInformation($"Got result: {result}");
await Task.Delay(1000, stoppingToken);
}
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{DFF091CA-D53E-4852-8742-A986136407EE}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{7566EA8A-D308-4B97-BBDC-4759C1275777}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{DFF091CA-D53E-4852-8742-A986136407EE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{DFF091CA-D53E-4852-8742-A986136407EE}.Debug|Any CPU.Build.0 = Debug|Any CPU
{DFF091CA-D53E-4852-8742-A986136407EE}.Release|Any CPU.ActiveCfg = Release|Any CPU
{DFF091CA-D53E-4852-8742-A986136407EE}.Release|Any CPU.Build.0 = Release|Any CPU
{7566EA8A-D308-4B97-BBDC-4759C1275777}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{7566EA8A-D308-4B97-BBDC-4759C1275777}.Debug|Any CPU.Build.0 = Debug|Any CPU
{7566EA8A-D308-4B97-BBDC-4759C1275777}.Release|Any CPU.ActiveCfg = Release|Any CPU
{7566EA8A-D308-4B97-BBDC-4759C1275777}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {F50157D4-E71D-4583-AD8D-5DA2EE3F792C}
EndGlobalSection
EndGlobal

Просмотреть файл

@ -0,0 +1,470 @@
# RabbitMQ Tutorial - Reliable Publishing with Publisher Confirms
## Publisher Confirms (using Steeltoe)
> #### Prerequisites
> This tutorial assumes RabbitMQ is [downloaded](https://www.rabbitmq.com/download.html) and installed and running
> on `localhost` on the [standard port](https://www.rabbitmq.com/networking.html#ports) (`5672`).
>
> In case you use a different host, port or credentials, connections settings would require adjusting.
>
> #### Where to get help
> If you're having trouble going through this tutorial you can contact us through Github issues on our
> [Steeltoe Samples Repository](https://github.com/SteeltoeOSS/Samples).
[Publisher confirms](https://www.rabbitmq.com/confirms.html#publisher-confirms)
are a RabbitMQ extension to implement reliable
publishing. When publisher confirms are enabled on a channel,
messages the client publishes are confirmed asynchronously
by the broker, meaning they have been taken care of on the server
side.
### Overview
In this tutorial we're going to use publisher confirms to make
sure published messages have safely reached the broker. We will cover several strategies to using publisher confirms and explain their pros and cons.
### Enabling Publisher Confirms on a Channel
Publisher confirms are a RabbitMQ extension to the AMQP 0.9.1 protocol,
so they are not enabled by default. Publisher confirms are
enabled at the channel level of a connection to the RabbitMQ broker.
Remember from the first tutorial we explained that Steeltoe adds to the service container a Caching zvonnection Factory that is used to create and cache connections to the RabbitMQ broker. By default, all of the Steeltoe RabbitMQ components (e.g. RabbitTemplate) use the factory when interacting with the broker (i.e. creating connections and channels).
By default the factory does not create connections/channels that have publisher confirms enabled. So in order to use publisher confirms in Steeltoe we need to add an additional connection factory to the service container configured with publisher confirms enabled. And we also then need to add an additional `RabbitTemplate` that is configured to use this additional connection factory.
We do that when configuring the services in the `RabbitMQHost` as follows:
```csharp
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Connection;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
internal const string QueueName = "hello";
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
// Add a connection factory with the name "publisherConfirmReturnsFactory"
// and configure it to use correlated publisher confirms
services.AddRabbitConnectionFactory("publisherConfirmReturnsFactory", (p, ccf) =>
{
ccf.IsPublisherReturns = true;
ccf.PublisherConfirmType = CachingConnectionFactory.ConfirmType.CORRELATED;
});
// Add an additional RabbitTemplate with the name "confirmReturnsTemplate"
// and configure it to use the above connection factory and mandatory delivery
services.AddRabbitTemplate("confirmReturnsTemplate", (p, template) =>
{
var ccf = p.GetRabbitConnectionFactory("publisherConfirmReturnsFactory");
template.ConnectionFactory = ccf;
template.Mandatory = true;
});
services.AddHostedService<Tut7Sender>();
})
.Build()
.Run();
}
}
}
```
Once this is done, then in our Sender we can use the following code to obtain the named `RabbitTemplate` which we configured properly above.
```csharp
namespace Sender
{
public class Tut7Sender : BackgroundService, IReturnCallback, IConfirmCallback
{
private readonly ILogger<Tut7Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
public Tut7Sender(ILogger<Tut7Sender> logger, IServiceProvider provider)
{
_logger = logger;
_rabbitTemplate = provider.GetRabbitTemplate("confirmReturnsTemplate");
....
}
}
}
```
### Strategy #1: Publishing Messages Individually
Let's start with the simplest approach to publishing with confirms,
that is, publishing a message and waiting synchronously for its confirmation:
```csharp
while (ThereAreMessagesToPublish()) {
....
_rabbitTemplate.ConvertAndSend(QueueName, (object)"Hello World!");
// Wait up to 5 seconds for confirmation
_rabbitTemplate.WaitForConfirmsOrDie(5000);
}
```
In the previous example we publish a message as usual and wait for its
confirmation with the `Channel#WaitForConfirmsOrDie(long)` method.
The method returns as soon as the message has been confirmed. If the
message is not confirmed within the timeout or if it is nack-ed (meaning
the broker could not take care of it for some reason), the method will
throw an exception. The handling of the exception usually consists
in logging an error message and/or retrying to send the message.
This technique is very straightforward but also has a major drawback:
it **significantly slows down publishing**, as the confirmation of a message blocks the publishing
of all subsequent messages. This approach is not going to deliver throughput of
more than a few hundreds of published messages per second. Nevertheless, this can be
good enough for some applications.
> #### Are Publisher Confirms Asynchronous?
>
> We mentioned at the beginning that the broker confirms published
> messages asynchronously but in the first example the code waits
> synchronously until the message is confirmed. The client actually
> receives confirms asynchronously and unblocks the call to `WaitForConfirmsOrDie`
> accordingly. Think of `WaitForConfirmsOrDie` as a synchronous helper
> which relies on asynchronous notifications under the hood.
### Strategy #2: Publishing Messages in Batches
To improve upon our previous example, we can publish a batch
of messages and wait for this whole batch to be confirmed.
The following example uses a batch of 100:
```csharp
int batchSize = 100;
int outstandingMessageCount = 0;
while (thereAreMessagesToPublish()) {
_rabbitTemplate.ConvertAndSend(QueueName, (object)"Hello World!");
outstandingMessageCount++;
if (outstandingMessageCount == batchSize) {
_rabbitTemplate.WaitForConfirmsOrDie(5000);
outstandingMessageCount = 0;
}
}
if (outstandingMessageCount > 0) {
_rabbitTemplate.WaitForConfirmsOrDie(5000);
}
```
Waiting for a batch of messages to be confirmed improves throughput drastically over
waiting for a confirm for individual message (up to 20-30 times with a remote RabbitMQ node).
One drawback is that we do not know exactly what went wrong in case of failure,
so we may have to keep a whole batch in memory to log something meaningful or
to re-publish the messages. And this solution is still synchronous, so it
blocks the publishing of messages.
### Strategy #3: Handling Publisher Confirms Asynchronously
The broker confirms published messages asynchronously, one just needs
to register a callback with the template to be notified of these confirms or returns.
Here is a simple example of how to do that:
```csharp
using Steeltoe.Messaging;
using Steeltoe.Messaging.RabbitMQ.Connection;
using Steeltoe.Messaging.RabbitMQ.Core;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using static Steeltoe.Messaging.RabbitMQ.Core.RabbitTemplate;
namespace Sender
{
public class Tut7Sender : BackgroundService, IReturnCallback, IConfirmCallback
{
private readonly ILogger<Tut7Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int id;
public Tut7Sender(ILogger<Tut7Sender> logger, IServiceProvider provider)
{
_logger = logger;
_rabbitTemplate = provider.GetRabbitTemplate("confirmReturnsTemplate");
_rabbitTemplate.ReturnCallback = this;
_rabbitTemplate.ConfirmCallback = this;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
CorrelationData data = new CorrelationData(id.ToString());
id++;
await _rabbitTemplate.ConvertAndSendAsync(Program.QueueName, (object)"Hello World!", data);
_logger.LogInformation("Worker running at: {time}, sent ID: {id}", DateTimeOffset.Now, data.Id);
await Task.Delay(1000, stoppingToken);
}
}
public void ReturnedMessage(IMessage<byte[]> message, int replyCode, string replyText, string exchange, string routingKey)
{
_logger.LogInformation($"Returned message: ReplyCode={replyCode}, Exchange={exchange}, RoutingKey={routingKey}");
}
public void Confirm(CorrelationData correlationData, bool ack, string cause)
{
_logger.LogInformation($"Confirming message: Id={correlationData.Id}, Acked={ack}, Cause={cause}");
}
}
}
```
There are two callbacks: one for confirmed messages and one returned messages.
```csharp
.....
public void ReturnedMessage(IMessage<byte[]> message, int replyCode, string replyText, string exchange, string routingKey)
{
_logger.LogInformation($"Returned message: ReplyCode={replyCode}, Exchange={exchange}, RoutingKey={routingKey}");
}
public void Confirm(CorrelationData correlationData, bool ack, string cause)
{
_logger.LogInformation($"Confirming message: Id={correlationData.Id}, Acked={ack}, Cause={cause}");
}
....
```
For the `ReturnedMessage()` callback method to be
invoked the templates property `Mandatory` must be set to true and the underlying connection factory must be configured
with `IsPublisherReturns` set to true. If those values are set, then the template will issue the returns callback to whatever is registered
with the template property `ReturnCallback`.
For publisher confirms (also known as publisher acknowledgements) to be enabled, the template requires the underlying connection factory
to have `PublisherConfirmType` property set to `ConfirmType.CORRELATED`. Then the template will issue confirm callbacks to whatever is registered
with the template property `ConfirmCallback.`
Note that the `CorrelationData` provided in the `Confirm(CorrelationData correlationData, ...)` is provided the user (developer) on the `ConvertAndSendAsync(...)` method call.
The template then returns it as part of the arguments to the `Confirm(...)` callback.
```csharp
....
CorrelationData data = new CorrelationData(id.ToString());
id++;
await _rabbitTemplate.ConvertAndSendAsync(Program.QueueName, (object)"Hello World!", data);
...
```
A simple way to correlate messages with sequence numbering consists in using a
dictionary of `CorrelationData` and messages . The publishing code can then track outbound
messages using the dictionary and upon receiving the `Confirm` callback can behave accordingly
depending on whether the message was acked or nacked.
### Summary
Making sure published messages made it to the broker can be essential in some applications.
Publisher confirms are a RabbitMQ feature that helps to meet this requirement. Publisher
confirms are asynchronous in nature but it is also possible to handle them synchronously.
There is no definitive way to implement publisher confirms, this usually comes down
to the constraints in the application and in the overall system. Typical techniques are:
* publishing messages individually, waiting for the confirmation synchronously: simple, but very
limited throughput.
* publishing messages in batch, waiting for the confirmation synchronously for a batch: simple, reasonable
throughput, but hard to reason about when something goes wrong.
* asynchronous handling: best performance and use of resources, good control in case of error, but
can be involved to implement correctly.
## Putting It All Together
The code for the receivers `Program.cs` comes from the first tutorial:
```csharp
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
internal const string QueueName = "hello";
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
// Add the rabbit listener
services.AddSingleton<Tut7Receiver>();
services.AddRabbitListeners<Tut7Receiver>();
})
.Build()
.Run();
}
}
}
```
The code for the `Tut7Receiver` also looks the same as in tutorial one:
```csharp
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
namespace Receiver
{
internal class Tut7Receiver
{
private readonly ILogger _logger;
public Tut7Receiver(ILogger<Tut7Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = Program.QueueName)]
public void Receive(string input)
{
_logger.LogInformation($"Received: {input}");
}
}
}
```
The code for the Sender is also based on tutorial one, but has the modifications for confirms.
The senders `Program.cs` is as follows:
```csharp
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Connection;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
internal const string QueueName = "hello";
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
services.AddRabbitConnectionFactory("publisherConfirmReturnsFactory", (p, ccf) =>
{
ccf.IsPublisherReturns = true;
ccf.PublisherConfirmType = CachingConnectionFactory.ConfirmType.CORRELATED;
});
services.AddRabbitTemplate("confirmReturnsTemplate", (p, template) =>
{
var ccf = p.GetRabbitConnectionFactory("publisherConfirmReturnsFactory");
template.ConnectionFactory = ccf;
template.Mandatory = true;
});
services.AddHostedService<Tut7Sender>();
})
.Build()
.Run();
}
}
}
```
The senders background service looks as follows:
```csharp
using Steeltoe.Messaging;
using Steeltoe.Messaging.RabbitMQ.Connection;
using Steeltoe.Messaging.RabbitMQ.Core;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using static Steeltoe.Messaging.RabbitMQ.Core.RabbitTemplate;
namespace Sender
{
public class Tut7Sender : BackgroundService, IReturnCallback, IConfirmCallback
{
private readonly ILogger<Tut7Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int id;
public Tut7Sender(ILogger<Tut7Sender> logger, IServiceProvider provider)
{
_logger = logger;
_rabbitTemplate = provider.GetRabbitTemplate("confirmReturnsTemplate");
_rabbitTemplate.ReturnCallback = this;
_rabbitTemplate.ConfirmCallback = this;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
CorrelationData data = new CorrelationData(id.ToString());
id++;
await _rabbitTemplate.ConvertAndSendAsync(Program.QueueName, (object)"Hello World!", data);
_logger.LogInformation("Worker running at: {time}, sent ID: {id}", DateTimeOffset.Now, data.Id);
await Task.Delay(1000, stoppingToken);
}
}
public void ReturnedMessage(IMessage<byte[]> message, int replyCode, string replyText, string exchange, string routingKey)
{
_logger.LogInformation($"Returned message: ReplyCode={replyCode}, Exchange={exchange}, RoutingKey={routingKey}");
}
public void Confirm(CorrelationData correlationData, bool ack, string cause)
{
_logger.LogInformation($"Confirming message: Id={correlationData.Id}, Acked={ack}, Cause={cause}");
}
}
}
```
Compile as usual, see [tutorial one](../Tutorial1/Readme.md)
```bash
cd tutorials\tutorial7
dotnet build
```
To run the receiver, execute the following commands:
```bash
# receiver
cd receiver
dotnet run
```
To watch the confirms come back, run the sender
```bash
# sender
cd sender
dotnet run
```

Просмотреть файл

@ -0,0 +1,28 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Receiver
{
internal class Program
{
internal const string QueueName = "hello";
static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices((hostContext, services) =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
// Add the rabbit listener
services.AddSingleton<Tut7Receiver>();
services.AddRabbitListeners<Tut7Receiver>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,26 @@
using Microsoft.Extensions.Logging;
using Steeltoe.Messaging.RabbitMQ.Attributes;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Receiver
{
internal class Tut7Receiver
{
private readonly ILogger _logger;
public Tut7Receiver(ILogger<Tut7Receiver> logger)
{
_logger = logger;
}
[RabbitListener(Queue = Program.QueueName)]
public void Receive(string input)
{
_logger.LogInformation($"Received: {input}");
}
}
}

Просмотреть файл

@ -0,0 +1,37 @@
using Steeltoe.Messaging.RabbitMQ.Config;
using Steeltoe.Messaging.RabbitMQ.Connection;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using Steeltoe.Messaging.RabbitMQ.Host;
namespace Sender
{
public class Program
{
internal const string QueueName = "hello";
public static void Main(string[] args)
{
RabbitMQHost.CreateDefaultBuilder(args)
.ConfigureServices(services =>
{
// Add queue to service container to be declared
services.AddRabbitQueue(new Queue(QueueName));
services.AddRabbitConnectionFactory("publisherConfirmReturnsFactory", (p, ccf) =>
{
ccf.IsPublisherReturns = true;
ccf.PublisherConfirmType = CachingConnectionFactory.ConfirmType.CORRELATED;
});
services.AddRabbitTemplate("confirmReturnsTemplate", (p, template) =>
{
var ccf = p.GetRabbitConnectionFactory("publisherConfirmReturnsFactory");
template.ConnectionFactory = ccf;
template.Mandatory = true;
});
services.AddHostedService<Tut7Sender>();
})
.Build()
.Run();
}
}
}

Просмотреть файл

@ -0,0 +1,11 @@
{
"profiles": {
"Sender": {
"commandName": "Project",
"dotnetRunMessages": true,
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

Просмотреть файл

@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Worker">
<PropertyGroup>
<TargetFramework>net6.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UserSecretsId>dotnet-Sender-C5FF4FBB-4AD5-435A-9D92-2A519E3770A0</UserSecretsId>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.1" />
<PackageReference Include="Steeltoe.Messaging.RabbitMQ" Version="3.2.1" />
</ItemGroup>
</Project>

Просмотреть файл

@ -0,0 +1,45 @@
using Steeltoe.Messaging;
using Steeltoe.Messaging.RabbitMQ.Connection;
using Steeltoe.Messaging.RabbitMQ.Core;
using Steeltoe.Messaging.RabbitMQ.Extensions;
using static Steeltoe.Messaging.RabbitMQ.Core.RabbitTemplate;
namespace Sender
{
public class Tut7Sender : BackgroundService, IReturnCallback, IConfirmCallback
{
private readonly ILogger<Tut7Sender> _logger;
private readonly RabbitTemplate _rabbitTemplate;
private int id;
public Tut7Sender(ILogger<Tut7Sender> logger, IServiceProvider provider)
{
_logger = logger;
_rabbitTemplate = provider.GetRabbitTemplate("confirmReturnsTemplate");
_rabbitTemplate.ReturnCallback = this;
_rabbitTemplate.ConfirmCallback = this;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
CorrelationData data = new CorrelationData(id.ToString());
id++;
await _rabbitTemplate.ConvertAndSendAsync(Program.QueueName, (object)"Hello World!", data);
_logger.LogInformation("Worker running at: {time}, sent ID: {id}", DateTimeOffset.Now, data.Id);
await Task.Delay(1000, stoppingToken);
}
}
public void ReturnedMessage(IMessage<byte[]> message, int replyCode, string replyText, string exchange, string routingKey)
{
_logger.LogInformation($"Returned message: ReplyCode={replyCode}, Exchange={exchange}, RoutingKey={routingKey}");
}
public void Confirm(CorrelationData correlationData, bool ack, string cause)
{
_logger.LogInformation($"Confirming message: Id={correlationData.Id}, Acked={ack}, Cause={cause}");
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

Просмотреть файл

@ -0,0 +1,31 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 17
VisualStudioVersion = 17.2.32901.213
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Receiver", "Receiver\Receiver.csproj", "{3C800E2C-A68F-4BFB-8EAB-80F285316A64}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sender", "Sender\Sender.csproj", "{328AAA6D-BC38-4496-AE5E-BC6DEA6668A1}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Release|Any CPU = Release|Any CPU
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{3C800E2C-A68F-4BFB-8EAB-80F285316A64}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{3C800E2C-A68F-4BFB-8EAB-80F285316A64}.Debug|Any CPU.Build.0 = Debug|Any CPU
{3C800E2C-A68F-4BFB-8EAB-80F285316A64}.Release|Any CPU.ActiveCfg = Release|Any CPU
{3C800E2C-A68F-4BFB-8EAB-80F285316A64}.Release|Any CPU.Build.0 = Release|Any CPU
{328AAA6D-BC38-4496-AE5E-BC6DEA6668A1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{328AAA6D-BC38-4496-AE5E-BC6DEA6668A1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{328AAA6D-BC38-4496-AE5E-BC6DEA6668A1}.Release|Any CPU.ActiveCfg = Release|Any CPU
{328AAA6D-BC38-4496-AE5E-BC6DEA6668A1}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {D637D0C7-B82B-4282-9D3A-F5A2A3059044}
EndGlobalSection
EndGlobal

Двоичные данные
Messaging/src/Tutorials/img/tutorials/VS2022NewConsoleApp.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 45 KiB

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 18 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/VS2022NewWorkService.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 40 KiB

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 18 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/VS2022Solution.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 36 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/bindings.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 5.6 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/consumer.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 2.1 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/direct-exchange-multiple.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 9.5 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/direct-exchange.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 10 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/exchanges.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 5.4 KiB

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 28 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/intro/exchange-declare.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 38 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/intro/exchange-direct.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 120 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/intro/exchange-fanout.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 73 KiB

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 40 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/intro/queue-declare-ok.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 40 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/intro/queue-declare.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 33 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/prefetch-count.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 8.2 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/producer.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 1.5 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/python-five.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 11 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/python-four.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 12 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/python-one-overall.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 4.2 KiB

Двоичные данные
Messaging/src/Tutorials/img/tutorials/python-one.png Normal file

Двоичный файл не отображается.

После

Ширина:  |  Высота:  |  Размер: 5.0 KiB

Некоторые файлы не были показаны из-за слишком большого количества измененных файлов Показать больше