我有一个使用 RabbitMQ 客户端接收消息的 .net 微服务，我需要测试以下内容：
1-消费者成功连接到rabbitMq主机。
2- 消费者正在监听队列。
3- 消费者接收消息成功。
为了实现上述目的，我创建了一个发送消息的示例应用程序，并且我正在调试消费者以确保它正在接收消息。
如何自动化这个测试？因此将其包含在我的微服务 CI 中。
我正在考虑将我的示例应用程序包含在我的 CI 中，这样我就可以触发一条消息，然后运行一个消费者单元测试，该测试等待特定时间然后在收到消息时通过，但这对我来说似乎是一种错误的做法，因为测试直到消息被触发几秒钟后才会开始。
我正在考虑的另一种方法是从单元测试本身触发示例应用程序，但如果示例应用程序无法运行，则会导致服务故障。
对通过 RabbitMQ 连接的微服务进行集成测试有什么最佳实践吗？
【问题讨论】：
标签： rabbitmq integration-testing microservices
我已经建立了许多这样的测试。我已经抛出了一些基本的代码 GitHub here with .NET Core 2.0.
您将需要一个 RabbitMQ 集群来进行这些自动化测试。每个测试都从消除队列开始，以确保不存在任何消息。来自另一个测试的预先存在的消息将中断当前测试。
我有一个简单的助手来删除队列。在我的应用程序中，它们总是声明自己的队列，但如果不是您的情况，那么您将不得不再次创建队列并绑定到任何交换。
public class QueueDestroyer
{
public static void DeleteQueue(string queueName, string virtualHost)
{
var connectionFactory = new ConnectionFactory();
connectionFactory.HostName = "localhost";
connectionFactory.UserName = "guest";
connectionFactory.Password = "guest";
connectionFactory.VirtualHost = virtualHost;
var connection = connectionFactory.CreateConnection();
var channel = connection.CreateModel();
channel.QueueDelete(queueName);
connection.Close();
}
}
我创建了一个非常简单的消费者示例来代表您的微服务。它在任务中运行，直到取消。
public class Consumer
{
private IMessageProcessor _messageProcessor;
private Task _consumerTask;
public Consumer(IMessageProcessor messageProcessor)
{
_messageProcessor = messageProcessor;
}
public void Consume(CancellationToken token, string queueName)
{
_consumerTask = Task.Run(() =>
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: queueName,
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
_messageProcessor.ProcessMessage(message);
};
channel.BasicConsume(queue: queueName,
autoAck: false,
consumer: consumer);
while (!token.IsCancellationRequested)
Thread.Sleep(1000);
}
}
});
}
public void WaitForCompletion()
{
_consumerTask.Wait();
}
}
消费者有一个 IMessageProcessor 接口来处理消息。在我的集成测试中，我创建了一个假的。您可能会为此使用您喜欢的模拟框架。
测试发布者向队列发布消息。
public class TestPublisher
{
public void Publish(string queueName, string message)
{
var factory = new ConnectionFactory() { HostName = "localhost", UserName="guest", Password="guest" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
routingKey: queueName,
basicProperties: null,
body: body);
}
}
}
我的示例测试如下所示：
[Fact]
public void If_SendMessageToQueue_ThenConsumerReceiv4es()
{
// ARRANGE
QueueDestroyer.DeleteQueue("queueX", "/");
var cts = new CancellationTokenSource();
var fake = new FakeProcessor();
var myMicroService = new Consumer(fake);
// ACT
myMicroService.Consume(cts.Token, "queueX");
var producer = new TestPublisher();
producer.Publish("queueX", "hello");
Thread.Sleep(1000); // make sure the consumer will have received the message
cts.Cancel();
// ASSERT
Assert.Equal(1, fake.Messages.Count);
Assert.Equal("hello", fake.Messages[0]);
}
我的假货是这样的：
public class FakeProcessor : IMessageProcessor
{
public List<string> Messages { get; set; }
public FakeProcessor()
{
Messages = new List<string>();
}
public void ProcessMessage(string message)
{
Messages.Add(message);
}
}
其他建议是：
如果您可以将随机文本附加到队列并在每次测试运行时交换名称，那么请这样做以避免并发测试相互干扰
如果您的应用程序不这样做，我在代码中还有一些帮助程序用于声明队列、交换和绑定。
编写一个连接杀手类，它将强制关闭连接并检查您的应用程序是否仍然工作并且可以恢复。我有这方面的代码，但在 .NET Core 中没有。只要问我，我可以修改它以在 .NET Core 中运行。
总的来说，我认为您应该避免在集成测试中包含其他微服务。例如，如果您将消息从一个服务发送到另一个服务并期望返回消息，则创建一个可以模拟预期行为的假消费者。如果您收到来自其他服务的消息，请在您的集成测试项目中创建虚假发布者。
【讨论】：
我成功地进行了这样的测试。您需要 RabbitMQ 的测试实例，测试交换以发送消息并测试队列以连接以接收消息。
不要嘲笑一切！
但是，对于 rabbitMQ 的测试消费者、生产者和测试实例，该测试中没有实际的生产代码。
使用测试 rabbitMQ 实例和实际应用
为了进行有意义的完整测试，我将使用测试 RabbitMQ 实例、交换和队列，但保留真正的应用程序（生产者和消费者）。
我会实现以下场景
当测试应用程序对rabbitMQ进行测试消息时
那么rabbitMQ中接收到的消息数量会增加
应用程序在收到消息时会做一些它应该做的事情
第 1 步和第 3 步是特定于应用程序的。您的应用程序根据一些外部事件（收到 HTTP 消息？定时器事件？）向 rabbitMQ 发送消息。您可以在测试中重现这种情况，因此应用程序将发送消息（测试 rabbitMQ 实例）。
在收到消息时验证应用程序操作的情况相同。应用程序应该在接收到消息时做一些可观察的事情。 如果应用程序进行 HTTP 调用，那么您可以模拟该 HTTP 端点并验证收到的消息。如果应用程序将消息保存到数据库 - 您可以将数据库池化以查找您的消息。
使用 rabbitMQ 监控 API
第 2 步可以使用 RabbitMQ 监控 API 实现（有一些方法可以查看从队列 @987654321@ 接收和消费的消息数量）
考虑使用 Spring Boot 进行健康检查
如果您是基于 java 的，那么使用 Spring Boot 将大大简化您的问题。您将自动获得 rabbitMQ 连接的健康检查！
有关如何使用 Spring boot 连接到 RabbitMQ 的教程，请参阅 https://spring.io/guides/gs/messaging-rabbitmq/。 Spring Boot 应用程序为每个附加的外部资源（数据库、消息传递、jms 等）公开健康信息（使用 HTTP 端点 /health） 详情请见https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-endpoints.html#_auto_configured_healthindicators。
如果与 rabbitMQ 的连接断开，则运行状况检查（由 org.springframework.boot.actuate.amqp.RabbitHealthIndicator 完成）将返回 HTTP 代码 4xx 和 JSON 正文中的 meaninfull json 消息。
你不需要做任何特别的事情来进行健康检查——只需使用 org.springframework.boot:spring-boot-starter-amqp 作为 maven/gradle 依赖就足够了。
CI 测试-来自 src/test 目录
我在 src/test 目录中使用集成测试编写了这样的测试（连接到 RabbitMQ 的外部测试实例）。如果使用 Spring Boot，最简单的方法是使用测试配置文件，并在 application-test.properties 中包含与测试 RabbitMQ 实例的连接详细信息（生产可以使用生产配置文件，以及带有 RabbitMQ 生产实例的 application-production.properties 文件）。
在最简单的情况下（只需验证与 rabbitMQ 的连接），您只需正常启动应用程序并验证 /health 端点。
在这种情况下，我会按照 CI 步骤进行操作
CI 测试 - 外部
上述方法也可以用于部署到测试环境（并连接到测试 rabbitMQ 实例）的应用程序。应用程序启动后，您可以检查 /health 端点以确保它已连接到 rabbitMQ 实例。
如果您让您的应用程序向rabbitMQ发送消息，那么您可以观察rabbbitMQ指标（使用rabbitMQ监控API）并观察应用程序消耗消息的外部影响。
对于此类测试，您需要在开始测试之前从 CI 启动和部署您的应用程序。
对于这种情况，我会按照 CI 步骤进行操作
考虑码头化环境
对于外部测试，您可以在 Docker 中运行您的应用程序以及测试 RabbitMQ 实例。您将需要两个 docker 容器。
要运行这两个镜像，写 docker-compose 文件是最合理的。
【讨论】：