选择消息代理：Kafka、RabbitMQ 还是 AWS SQS/SNS
微服务应用严重依赖消息传递和异步通信来保持一切顺利运行。
在开发需要相互通信的服务时，选择合适的消息代理是您必须做出的首要关键选择之一。
做出“正确”的选择可能是在各种功能和特殊情况之间进行权衡，而这些功能和特殊情况可能很难区分。
在本文中，我将概述一些比较知名的消息代理——Kafka、RabbitMQ 和 AWS SQS/SNS，以提供一些指导意义。
我将探讨它们背后的驱动力、它们遵循的一般信息传递模式，并尽我所能为选择适合您的经纪商提供一些指导。
Apache Kafka
Kafka是一个开源消息代理，主要由Apache 软件基金会开发和维护，并得到了开源社区的协助。
主要特点
- 专注于可流式内容，处理大型数据流
- 消息持久性和重新处理能力是核心功能
- 提供第三方选项的本地托管
Kafka 提供基于流的优化事件处理，采用发布/订阅模型驱动消费者之间的通信。
这些事件可以细分为主题，从而更好地组织分布式应用程序的通信模式，并被划分到集群内的多个服务器上，从而实现弹性和高性能的消息传递系统。
技术细节和部署
Apache提供多种不同语言的SDK 。
Kafka 的设计初衷是部署在本地，集成到您自己的应用程序架构中。它可以部署在一组独立的服务器上、虚拟机上，或者 Docker 容器中。
多家供应商提供 Kafka 托管服务，例如AWS、CloudKarafka和Aiven，或者在虚拟机中提供 Kafka 托管服务。
以下是一些用于入门Apache Kafka 事件的示例 JS 代码：
const { Kafka } = require('kafkajs')
const kafka = new Kafka({
clientId: 'my-app',
brokers: ['localhost:9092']
})
// this produces a message
async function produce() {
const producer = kafka.producer()
await producer.connect()
await producer.send({
topic: 'TOPIC_NAME',
messages: [
{ key: 'key1', value: 'hello world' },
],
})
}
async function consume() {
const consumer = kafka.consumer({ groupId: 'my-group' })
await consumer.connect()
await consumer.subscribe({ topic: 'TOPIC_NAME' })
await consumer.run({
eachMessage: async ({ topic, partition, message }) => {
console.log({
key: message.key.toString(),
value: message.value.toString(),
headers: message.headers,
})
},
})
}
优势与劣势
Kafka 非常注重数据流吞吐量，这一点在其性能统计数据中有所体现。
这种专注于处理数据流的方式，使得系统具有高吞吐量，从而能够对大型数据流进行复杂处理。
与其他消息代理相比，Kafka 对这些数据流的路由能力相对有限——随着这些产品的改进，这种差距正在不断缩小。
总而言之，Kafka 是一个强大的解决方案，可以提供强大且容错的高性能消息流，让您能够自信地控制应用程序的行为。
根据您的带宽和资源，您可以根据自己的需要选择抽象化尽可能多的或最少的托管，这使得 Kafka 成为一个可靠的选择，可以随着您的流量而扩展。
RabbitMQ
与 Kafka 类似，RabbitMQ也是一款开源消息代理。该技术最初由 Rabbit Technologies 开发，后经一系列收购，最终归 VMware 所有。
主要特点
- 专注于基于消息传递的通信，并支持大数据流
- 提供复杂的路由功能作为核心特性
- 提供第三方选项的本地托管
RabbitMQ 也使用发布/订阅模型，以二进制形式将消息对象发送到不同的命名队列，这些队列可以动态创建和销毁。
RabbitMQ 既可以独立运行，也可以作为集群的一部分运行，提供足够的配置能力来满足任何冗余或数据安全需求。
技术细节和部署
RabbitMQ 提供了多种语言的客户端库。
它可以部署在本地，从完整的服务器到容器，或者部署在多个云提供商之一：
以下示例代码使用 Node.js 和 AMQPLIB 包编写，可以简要展示使用 RabbitMQ 的体验：
const amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(error0, connection) {
if (error0) {
throw error0;
}
connection.createChannel(function(error1, channel) {
if (error1) {
throw error1;
}
const queue = 'hello-queue';
const msg = 'Hello world!';
channel.assertQueue(queue, {
durable: false
});
// Sending message to queue
channel.sendToQueue(queue, Buffer.from(msg));
console.log("Sent message", msg);
// Consuming messages
channel.consume(queue, function(msg) {
console.log("Received message", msg.content.toString());
}, { noAck: true });
});
});
优势与劣势
RabbitMQ 能够处理几乎任何规模的工作负载，并且可以随着用户群的增长而有效地与您的应用程序一起扩展。
RabbitMQ 专注于基于消息的传输和复杂的路由场景，因此对任何应用程序架构都具有极强的适应性。
虽然最初对数据流处理的支持力度不够，消息通常只处理一次，没有能力重新处理数据流，但随着 RabbitMQ 的不断发展，这两个差距都已得到弥补。
RabbitMQ 能够让你掌控自己想要掌控的部分，并将其余部分外包出去，因此它可以在你的应用程序基础设施中扮演任何合适的角色。
亚马逊网络服务 SQS/SNS
SNS 非常注重消息传递，提供发布-订阅模型，以便快速将消息分发给一系列客户端（例如，移动设备、HTTPS 端点、其他 AWS 服务）。
相比之下，SQS 则专注于各个客户端成功传递和处理消息。
主要特点
- 两款产品，均支持广播消息和发布/订阅功能
- 使用 AWS 快速设置和配置
- AWS 之外不提供任何托管服务。
SNS 会将同一条消息广播给一组接收者，而 SQS 会将队列消息分发给单个订阅者进行处理。
SNS 采用推送式通知方式，允许对通知活动进行自动响应，而 SQS 则更侧重于轮询式机制，并支持一些额外的事件驱动功能。
技术细节
AWS 提供了一个通用 SDK，支持多种流行语言，可访问大多数 AWS 服务（包括 SQS 和 SNS）。
以下示例代码使用 AWS SDK 来演示如何使用 SNS 和 SQS：
// SNS - publish
const AWS = require('aws-sdk');
AWS.config.update({ region: 'REGION' });
const publishParams = {
Message: 'MESSAGE_TEXT',
TopicArn: 'TOPIC_ARN'
};
const publishTextPromise = new AWS.SNS({ apiVersion: '2010-03-31' }).publish(publishParams).promise();
publishTextPromise.then(
function(data) {
console.log(`Message ${publishParams.Message} sent to topic ${publishParams.TopicArn}`);
}).catch(
function(err) {
console.error(err, err.stack);
});
// SNS - Subscribe
const subscribeParams = {
TopicArn : 'TOPIC_ARN'
}
const subscribePromise = new AWS.SNS({ apiVersion: '2010-03-31' }).listSubscriptionsByTopic(subscribeParams).promise();
subscribePromise.then(
function(data) {
console.log(data);
}).catch(
function(err) {
console.error(err, err.stack);
}
);
// SQS - send
const sqs = new AWS.SQS({ apiVersion: '2012-11-05' });
const queueURL = "SQS_QUEUE_URL";
const sendParams = {
DelaySeconds: 10,
MessageAttributes: {
"Title": {
DataType: "String",
StringValue: "Some String"
}
},
MessageBody: "Something!",
QueueUrl: queueURL
};
sqs.sendMessage(sendParams, function(err, data) {
if (err) {
console.log("Error sending to SQS", err);
} else {
console.log("Success sending to SQS", data.MessageId);
}
});
// SQS - receive
const receiveParams = {
AttributeNames: [
"SentTimestamp"
],
MaxNumberOfMessages: 10,
MessageAttributeNames: [
"All"
],
QueueUrl: queueURL,
VisibilityTimeout: 20,
WaitTimeSeconds: 0
};
sqs.receiveMessage(receiveParams, function(err, data) {
if (err) {
console.log("Receive Error", err);
} else if (data.Messages) {
console.log("Received messages:", JSON.stringify(data.Messages))
}
});
优势与劣势
AWS SQS 和 SNS 可以结合起来，构建高度可扩展、高度弹性的分布式应用程序的骨干网络。
这两个工具与许多其他 AWS 服务（例如 AWS Lambda）集成，可以帮助您轻松扩展应用程序的通信，同时为您提供管理服务交互底层复杂性所需的所有工具。
如果您的 Web 应用程序已在 AWS 上运行，则设置时间几乎为零，而且比许多其他系统简单得多。但随着消息数量的增长，这可能会增加 AWS 账单。
虽然 Kafka 和 RabbitMQ 没有提供默认的消息大小限制，但 AWS 对 SQS 和 SNS 消息提供了一些限制——在消息达到一定大小后将其转换为 S3 对象。
我们发表了一篇详细的文章，介绍如何克服这个大小限制——我强烈建议您浏览一下，以便了解 SQS 特别是如何处理大型消息。
[您可以在其中找到我们的 SQS/SNS 生产者/消费者库。它提供了通过 S3 传递有效负载的功能。]
SQS 和 SNS 作为云优先技术，确实增加了被供应商锁定到特定服务的复杂性，而其他消息代理则通过提供本地安装和维护功能来解决这个问题。
选择合适的消息代理
一般来说，选择经纪人时应考虑以下两点：
考虑因素一：您发送的消息类型
选择消息代理的第一步是确定你要发送哪些消息，以及它们的通用格式是什么。
这些消息的特点将决定我们需要对每个平台的产品提出哪些问题，尽管大多数平台在功能集方面大致相同——这意味着，从总体上看，上面列出的每个解决方案都支持作为可扩展分布式应用程序的消息代理所需的功能。
从纯粹的功能角度来看，这两个解决方案都一样好。
考虑因素二：您的日常工作和应用程序的基础架构
这就需要考虑一些次要因素了。想想你的日常工作和系统，然后问问自己：
- 您是否完全在 AWS 上构建应用程序？如果是这样，那么 SQS 和 SNS 或许是建立服务间通信的最佳选择。
- 您是否更专注于编写应用程序，而不是维护组件之间的数据管道？如果是这样，第三方托管解决方案可能是最佳选择，它可以让您专注于自身优势，同时扩展代码库。
- 如果您非常注重交付速度和最低延迟，那么 Kinesis 可能非常适合您（我们将在另一篇文章中详细介绍 Kinesis，敬请期待）；而更注重验证交付和冗余性的应用则可能需要选择其他技术。
在这个层面上，应用程序的基础设施和行为模式的需求将决定最终的选择。
考虑到以上因素，并且需要指出的是，将这些大型科技产品简化为几行建议是困难的（而且在某种程度上也不公平），以下是选择合适的消息代理的一些指导原则：
- 如果您重视消息保留和轻松重新处理数据的能力，Kafka 可能是您的最佳选择。
- 如果您更关注维护和实施复杂的路由规则集的能力，那么 RabbitMQ 可能是您的最佳选择。
- 如果您是一家希望快速启动并运行且成本最低的小型初创公司，鉴于其快速的设置和成本结构，AWS SQS/SNS 是一个绝佳的选择。
全面了解消息传递过程
你需要评估的一个要素是如何最好地维护最终产品。一旦你的应用程序开始发送消息，当出现问题时，我们该如何追踪问题所在？
OpenTelemetry是一套 SDK 和工具，可用于为您的分布式应用程序设置可观测性，为您提供在应用程序出现问题时排查分布式消息传递故障的手段。
以下是关于如何在分布式应用程序中实现 OpenTelemetry 的快速分步指南，它可以帮助您全面了解消息流转过程。本指南以 Kafka 作为消息代理为例进行演示。
[附注：请下载适用于 Node.js 的 OpenTelemetry Kafkajs Instrumentation。]
结论
如果你正在构建一个以任何方式“分布式”的应用程序，那么你很可能在某个时候需要处理应用程序组件之间的异步通信。
消息（以及传递消息的代理）将在驱动应用程序的基础架构中发挥关键作用。
以上总结绝非详尽无遗——我可能还需要再写一千字才能真正开始全面描述消息代理领域——但希望能提供一些有价值的信息，供您在做决定时参考。
关键在于充分了解您的应用程序的需求，以及这些需求如何与您正在评估的消息代理的功能相匹配。
最终选择哪个消息代理并没有“错误”答案，但希望以上信息能帮助您找到正确的方向。
About Aspecto
Aspecto is an OpenTelemetry-based troubleshooting platform
that helps developers prevent distributed application
issues from their local dev environment,
and across the entire development cycle.
You can think of it as the
Chrome DevTools for your distributed applications.
Aspecto is used for detecting and
troubleshooting microservices-based distributed systems,
and preventing software failures before deployment.
Visit us at Aspecto.io
for more microservices tutorials