大数据领域RabbitMQ与移动应用的数据交互

关键词：RabbitMQ、消息队列、移动应用、大数据、异步通信、数据交互、AMQP协议

摘要：本文深入探讨RabbitMQ在大数据领域与移动应用数据交互中的应用。我们将从基础概念出发，逐步解析RabbitMQ的核心原理，并通过实际案例展示如何构建高效可靠的移动应用数据交互系统。文章将涵盖RabbitMQ的架构设计、与移动应用的集成方式、性能优化策略以及在大数据环境下的最佳实践。

背景介绍

目的和范围

本文旨在为开发者和架构师提供RabbitMQ在大数据领域与移动应用数据交互的全面指南。我们将覆盖从基础概念到高级应用的完整知识体系，包括设计模式、性能调优和故障处理等关键主题。

预期读者

• 移动应用开发工程师
• 后端系统架构师
• 大数据工程师
• 消息中间件技术爱好者
• 希望了解异步通信机制的技术管理者

文档结构概述

文章首先介绍RabbitMQ的核心概念，然后深入探讨其与移动应用的集成方式，接着通过实际案例展示具体实现，最后讨论性能优化和未来发展趋势。

术语表

核心术语定义

• RabbitMQ：开源消息代理软件，实现了高级消息队列协议(AMQP)
• 消息队列：应用程序之间通信的方法，通过发送和接收消息实现解耦
• 生产者(Producer)：发送消息的应用程序
• 消费者(Consumer)：接收消息的应用程序
• 交换器(Exchange)：接收生产者发送的消息并根据规则路由到队列

相关概念解释

• AMQP协议：高级消息队列协议，RabbitMQ的核心协议
• 消息持久化：确保消息在服务器重启后不会丢失的机制
• 负载均衡：在多消费者情况下分配消息处理的策略
• 死信队列：处理无法被正常消费的消息的特殊队列

缩略词列表

• AMQP: Advanced Message Queuing Protocol
• MQTT: Message Queuing Telemetry Transport
• API: Application Programming Interface
• JSON: JavaScript Object Notation
• REST: Representational State Transfer

核心概念与联系

故事引入

想象你经营着一家大型外卖平台，每天有数百万的订单需要处理。移动应用用户下单后，系统需要通知餐厅准备食物、分配骑手配送、更新用户订单状态，还要记录数据用于分析。如果所有步骤都同步进行，就像让一个人同时接电话、做饭、开车送货和记账，肯定会手忙脚乱。RabbitMQ就像一位聪明的调度员，它把每个任务写成小纸条(消息)，分发给不同的工作人员(服务)，大家各司其职，整个系统就能高效运转。

核心概念解释

核心概念一：消息队列

消息队列就像一个邮局系统。当移动应用(发件人)需要发送数据时，它不直接联系接收方，而是把数据打包成信件(消息)交给邮局(RabbitMQ)。邮局负责把信件安全地送到正确的邮箱(队列)里，接收方(消费者)可以在自己方便的时候去邮箱取信。这样即使接收方暂时不在线，信件也不会丢失。

核心概念二：交换器和路由

RabbitMQ中的交换器就像邮局的分拣中心。当移动应用发送消息时，它首先到达交换器。交换器根据预定义的规则(路由键)决定把消息投递到哪个队列。就像邮局会根据邮政编码和地址把信件分到不同的邮递路线。

核心概念三：消息确认机制

为了确保消息不丢失，RabbitMQ采用了确认机制。当消费者成功处理消息后，会发送一个确认回执给RabbitMQ。如果RabbitMQ没有收到确认，它会认为消息处理失败，并可能重新投递。就像快递签收，只有收件人签字确认后，快递员才会认为任务完成。

核心概念之间的关系

消息队列与交换器的关系

消息队列和交换器就像邮局系统和分拣中心的关系。移动应用(生产者)把消息交给交换器(分拣中心)，交换器根据规则把消息路由到不同的队列(邮递路线)，最终由消费者(收件人)从队列中获取消息。

交换器与路由键的关系

路由键就像信封上的邮政编码和地址信息。交换器通过检查消息的路由键，决定如何分发消息。不同类型的交换器(直连、主题、扇出)使用不同的规则来解释路由键，就像不同的邮递服务(普通邮件、挂号信、快递)有不同的分拣标准。

消息确认与可靠性的关系

消息确认机制是保证系统可靠性的关键。没有确认机制，RabbitMQ就不知道消费者是否成功处理了消息，可能导致数据丢失或重复处理。这就像快递没有签收确认，发件人就不知道包裹是否安全送达。

核心概念原理和架构的文本示意图

[移动应用] -> (发布消息) -> [Exchange] 
    Exchange根据类型和路由键决定消息去向:
    - 直连交换器: 精确匹配路由键
    - 主题交换器: 模式匹配路由键
    - 扇出交换器: 广播到所有绑定队列
    -> [Queue] -> (消费消息) -> [大数据处理服务]

Mermaid流程图

核心算法原理 & 具体操作步骤

RabbitMQ的核心算法原理基于AMQP协议，下面我们通过Python代码示例来展示其工作流程。

生产者代码示例

import pika
import json

# 连接RabbitMQ服务器
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 声明一个直连交换器
channel.exchange_declare(exchange='mobile_app_events', 
                        exchange_type='direct')

# 准备移动应用事件数据
event = {
    "user_id": "12345",
    "event_type": "purchase",
    "timestamp": "2023-05-20T14:30:00Z",
    "product_id": "67890",
    "amount": 99.99
}

# 发布消息到交换器，指定路由键
channel.basic_publish(
    exchange='mobile_app_events',
    routing_key='purchase',  # 使用事件类型作为路由键
    body=json.dumps(event),
    properties=pika.BasicProperties(
        delivery_mode=2,  # 使消息持久化
    )
)

print(" [x] 发送事件 %r" % event)
connection.close()

消费者代码示例

import pika
import json

# 连接RabbitMQ服务器
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 声明相同的交换器
channel.exchange_declare(exchange='mobile_app_events',
                        exchange_type='direct')

# 创建匿名队列
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# 将队列绑定到交换器，只关注购买事件
channel.queue_bind(exchange='mobile_app_events',
                  queue=queue_name,
                  routing_key='purchase')

print(' [*] 等待购买事件。按 CTRL+C 退出')

# 定义回调函数处理消息
def callback(ch, method, properties, body):
    event = json.loads(body)
    print(" [x] 收到 %r" % event)
    # 在这里添加实际的事件处理逻辑
    # ...
    # 手动发送确认
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 设置公平分发
channel.basic_qos(prefetch_count=1)

# 开始消费消息
channel.basic_consume(queue=queue_name,
                     on_message_callback=callback)

channel.start_consuming()

操作步骤详解

1. 建立连接：生产者和消费者都需要与RabbitMQ服务器建立TCP连接。

2. 声明交换器：交换器是消息路由的核心，必须在使用前声明其类型和名称。

3. 准备队列：

   • 生产者通常不直接与队列交互
   • 消费者可以创建匿名临时队列或使用持久化队列

4. 绑定队列到交换器：指定哪些队列接收来自特定交换器的消息，以及基于什么路由键。

5. 发布消息：生产者将消息发送到交换器，并指定路由键。

6. 消费消息：消费者订阅队列并定义回调函数处理到达的消息。

7. 消息确认：消费者处理完消息后必须显式确认，确保可靠性。

数学模型和公式

RabbitMQ的性能可以通过以下数学模型进行分析：

消息吞吐量模型

RabbitMQ的消息吞吐量可以表示为：

$$T=\frac{N}{t_p+t_c+t_q}$$

其中：

• $T$: 系统吞吐量(消息/秒)
• $N$: 并发连接数
• $t_p$: 生产者发布消息的平均时间
• $t_c$: 消费者处理消息的平均时间
• $t_q$: 消息在队列中的等待时间

队列长度预测

使用Little’s Law预测队列长度：

$$L=\lambda W$$

其中：

• $L$: 队列中的平均消息数
• $\lambda$: 平均到达率(消息/秒)
• $W$: 消息在队列中的平均等待时间

负载均衡策略

在多消费者场景下，可以使用以下公式计算最优的预取值(prefetch count):

$$P_{opt}=\lceil \frac{t_{process}}{t_{network}}\rceil$$

其中：

• $P_{opt}$: 最优预取值
• $t_{process}$: 平均消息处理时间
• $t_{network}$: 网络往返延迟

项目实战：代码实际案例和详细解释说明

开发环境搭建

1. 安装RabbitMQ：

   # Ubuntu
   sudo apt-get install rabbitmq-server
   
   # MacOS
   brew install rabbitmq
   
   # Windows
   # 从官网下载安装包: https://www.rabbitmq.com/install-windows.html
   

2. 启动RabbitMQ服务：

   # Linux/MacOS
   sudo systemctl start rabbitmq-server
   
   # 或使用brew services
   brew services start rabbitmq
   

3. 安装Python客户端库：

   pip install pika
   

移动应用数据收集系统实现

下面我们实现一个完整的移动应用数据收集系统，包含以下组件：

1. 移动应用模拟器(生产者)
2. 事件处理服务(消费者)
3. 数据分析服务(消费者)

1. 移动应用模拟器(生产者)

import pika
import json
import random
import time
from datetime import datetime

# 模拟的用户行为事件类型
EVENT_TYPES = [
    'login', 'logout', 'purchase', 'view_item', 
    'add_to_cart', 'remove_from_cart', 'search'
]

# 连接RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明主题交换器，用于灵活路由
channel.exchange_declare(
    exchange='mobile_events',
    exchange_type='topic'
)

# 模拟用户事件生成
def generate_event():
    user_id = random.randint(1000, 9999)
    event_type = random.choice(EVENT_TYPES)
    timestamp = datetime.utcnow().isoformat() + "Z"
    
    # 根据不同事件类型生成特定数据
    if event_type == 'purchase':
        data = {
            'order_id': random.randint(100000, 999999),
            'amount': round(random.uniform(5, 200), 2),
            'items': [
                {'product_id': random.randint(1, 100), 
                 'quantity': random.randint(1, 5)}
                for _ in range(random.randint(1, 3))
            ]
        }
    elif event_type == 'view_item':
        data = {'product_id': random.randint(1, 100)}
    else:
        data = {}
    
    return {
        'user_id': user_id,
        'event_type': event_type,
        'timestamp': timestamp,
        'data': data
    }

# 持续生成并发送事件
try:
    while True:
        event = generate_event()
        
        # 发布到主题交换器，使用event_type作为路由键
        channel.basic_publish(
            exchange='mobile_events',
            routing_key=event['event_type'],
            body=json.dumps(event),
            properties=pika.BasicProperties(
                delivery_mode=2,  # 持久化消息
                content_type='application/json',
                headers={
                    'app_version': '1.2.3',
                    'device_type': random.choice(['ios', 'android'])
                }
            )
        )
        
        print(f" [x] 发送事件: {event['event_type']}")
        time.sleep(random.uniform(0.1, 0.5))
        
except KeyboardInterrupt:
    print("停止事件生成")
    connection.close()

2. 实时事件处理服务(消费者)

import pika
import json
import time

# 连接RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明相同的交换器
channel.exchange_declare(
    exchange='mobile_events',
    exchange_type='topic'
)

# 创建持久化队列
result = channel.queue_declare(
    queue='real_time_processing',
    durable=True
)

# 绑定关注的事件类型
for event_type in ['purchase', 'add_to_cart']:
    channel.queue_bind(
        exchange='mobile_events',
        queue='real_time_processing',
        routing_key=event_type
    )

print(' [*] 等待实时事件。按 CTRL+C 退出')

# 处理消息的回调函数
def callback(ch, method, properties, body):
    event = json.loads(body)
    
    print(f" [x] 处理 {method.routing_key} 事件:")
    print(f"      用户: {event['user_id']}")
    print(f"      时间: {event['timestamp']}")
    
    # 模拟处理时间
    time.sleep(0.3)
    
    # 确认消息处理完成
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(" [x] 处理完成")

# 设置公平分发
channel.basic_qos(prefetch_count=5)

# 开始消费
channel.basic_consume(
    queue='real_time_processing',
    on_message_callback=callback
)

channel.start_consuming()

3. 数据分析服务(消费者)

import pika
import json
import sqlite3
from collections import defaultdict

# 初始化SQLite数据库
def init_db():
    conn = sqlite3.connect('analytics.db')
    c = conn.cursor()
    
    # 创建事件统计表
    c.execute('''CREATE TABLE IF NOT EXISTS event_stats
                 (date TEXT, event_type TEXT, count INTEGER,
                  PRIMARY KEY (date, event_type))''')
    
    # 创建用户行为表
    c.execute('''CREATE TABLE IF NOT EXISTS user_behavior
                 (user_id INTEGER, event_type TEXT, count INTEGER,
                  last_event_time TEXT, PRIMARY KEY (user_id, event_type))''')
    
    conn.commit()
    conn.close()

# 连接RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明相同的交换器
channel.exchange_declare(
    exchange='mobile_events',
    exchange_type='topic'
)

# 创建持久化队列
result = channel.queue_declare(
    queue='analytics_processing',
    durable=True
)

# 绑定所有事件类型
channel.queue_bind(
    exchange='mobile_events',
    queue='analytics_processing',
    routing_key='#'  # #通配符匹配所有路由键
)

print(' [*] 等待分析事件。按 CTRL+C 退出')

# 初始化内存中的统计
daily_stats = defaultdict(lambda: defaultdict(int))
user_behavior = defaultdict(lambda: defaultdict(int))

def update_analytics(event):
    # 提取日期部分
    event_date = event['timestamp'].split('T')[0]
    event_type = event['event_type']
    user_id = event['user_id']
    
    # 更新内存统计
    daily_stats[event_date][event_type] += 1
    user_behavior[user_id][event_type] += 1
    
    # 每10个事件或特定事件类型时写入数据库
    if sum(daily_stats[event_date].values()) % 10 == 0 or event_type == 'purchase':
        save_to_db()

def save_to_db():
    conn = sqlite3.connect('analytics.db')
    c = conn.cursor()
    
    # 更新事件统计
    for date, stats in daily_stats.items():
        for event_type, count in stats.items():
            c.execute('''INSERT OR REPLACE INTO event_stats
                         (date, event_type, count)
                         VALUES (?, ?, COALESCE(
                             (SELECT count FROM event_stats 
                              WHERE date=? AND event_type=?), 0) + ?)''',
                     (date, event_type, date, event_type, count))
    
    # 更新用户行为
    for user_id, behaviors in user_behavior.items():
        for event_type, count in behaviors.items():
            c.execute('''INSERT OR REPLACE INTO user_behavior
                         (user_id, event_type, count, last_event_time)
                         VALUES (?, ?, COALESCE(
                             (SELECT count FROM user_behavior 
                              WHERE user_id=? AND event_type=?), 0) + ?, 
                             datetime('now'))''',
                     (user_id, event_type, user_id, event_type, count))
    
    conn.commit()
    conn.close()
    print(" [*] 分析数据已保存到数据库")

# 处理消息的回调函数
def callback(ch, method, properties, body):
    event = json.loads(body)
    
    # 更新分析数据
    update_analytics(event)
    
    # 确认消息处理完成
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 设置公平分发
channel.basic_qos(prefetch_count=10)

# 开始消费
channel.basic_consume(
    queue='analytics_processing',
    on_message_callback=callback
)

# 初始化数据库
init_db()

try:
    channel.start_consuming()
except KeyboardInterrupt:
    # 退出前保存剩余数据
    save_to_db()
    connection.close()

代码解读与分析

1. 生产者设计：

   • 使用主题交换器(topic exchange)实现灵活的路由
   • 为不同事件类型设置不同的路由键
   • 消息包含丰富的元数据(headers)
   • 实现消息持久化确保可靠性

2. 实时事件处理服务：

   • 只关注关键业务事件(购买和加购)
   • 使用持久化队列确保消息不丢失
   • 实现公平分发(prefetch_count=5)平衡负载
   • 显式确认机制保证消息正确处理

3. 数据分析服务：

   • 使用#通配符接收所有事件
   • 内存中聚合统计减少数据库写入
   • 批处理写入提高数据库性能
   • 使用SQLite作为轻量级分析存储

4. 性能考虑：

   • 不同消费者可以独立扩展
   • 根据处理能力设置不同的预取值
   • 持久化队列确保系统可靠性
   • 主题交换器提供灵活的路由能力

实际应用场景

场景一：用户行为分析

移动应用可以将用户的各种行为(点击、浏览、购买等)作为消息发送到RabbitMQ。数据分析服务可以实时消费这些消息，生成用户画像和行为分析报告。

优势：

• 解耦移动应用和数据分析系统
• 支持实时和批量处理
• 轻松应对流量高峰

场景二：订单处理系统

移动电商应用下单后，通过RabbitMQ将订单信息分发给不同的处理服务：库存管理、支付处理、物流调度等。

实现方式：

1. 移动应用发布订单消息到"orders"交换器
2. 库存服务订阅"orders.inventory"队列
3. 支付服务订阅"orders.payment"队列
4. 物流服务订阅"orders.shipping"队列

场景三：跨平台通知

在社交应用中，当用户发布新内容时，需要通知其粉丝的移动设备。RabbitMQ可以高效分发这些通知。

架构：

[移动应用] -> [内容发布] -> [RabbitMQ] -> [推送服务] -> [用户设备]
           -> [粉丝关系服务] -> [RabbitMQ] -> [推送服务] -> [用户设备]

场景四：离线数据同步

移动应用在离线状态下收集的数据，在网络恢复后可以通过RabbitMQ批量上传到服务器。

特点：

• 支持断点续传
• 自动重试失败的消息
• 优先级队列处理关键数据

工具和资源推荐

管理工具

1. RabbitMQ Management Plugin：官方提供的Web管理界面

   rabbitmq-plugins enable rabbitmq_management
   

   访问: http://localhost:15672 (默认凭据: guest/guest)

2. RabbitMQ CLI Tools：命令行管理工具

   # 列出队列
   rabbitmqadmin list queues
   
   # 查看绑定
   rabbitmqadmin list bindings
   

监控工具

1. Prometheus + Grafana：监控RabbitMQ指标

   • 使用rabbitmq_prometheus插件
   • 配置Grafana仪表板

2. Datadog/New Relic：商业监控解决方案

   • 提供开箱即用的RabbitMQ监控
   • 警报和自动化功能

客户端库

1. Python：pika

   pip install pika
   

2. Java：amqp-client

   <dependency>
     <groupId>com.rabbitmq</groupId>
     <artifactId>amqp-client</artifactId>
     <version>5.14.2</version>
   </dependency>
   

3. Node.js：amqplib

   npm install amqplib
   

学习资源

1. 官方文档：https://www.rabbitmq.com/documentation.html
2. RabbitMQ in Action：Alvaro Videla和Jason J.W. Williams著
3. RabbitMQ Patterns：https://www.rabbitmq.com/tutorials/amqp-concepts.html

未来发展趋势与挑战

趋势一：与云服务的深度集成

主要云平台(AWS, Azure, GCP)都提供了托管的RabbitMQ服务，简化了部署和运维。

优势：

• 自动扩展
• 内置监控
• 高可用性配置

趋势二：与Kafka的协同使用

在大数据场景下，RabbitMQ和Kafka可以互补使用：

• RabbitMQ：处理实时操作、低延迟消息
• Kafka：处理高吞吐量的事件流和长期存储

趋势三：物联网(IoT)应用

RabbitMQ的轻量级MQTT插件使其成为移动和IoT设备的理想选择。

应用场景：

• 移动设备状态监控
• 传感器数据收集
• 远程控制指令下发

挑战一：大规模部署

当消息量达到数百万/天时，需要考虑：

• 集群配置
• 镜像队列
• 网络分区处理

挑战二：消息顺序保证

RabbitMQ不保证全局消息顺序，需要应用层处理顺序敏感的场景。

解决方案：

• 单队列单消费者模式
• 消息版本控制
• 序列号检测

挑战三：移动网络不稳定性

移动设备网络连接不稳定，需要特殊处理：

• 持久化消息
• 自动重连机制
• 离线队列缓冲

总结：学到了什么？

核心概念回顾

1. RabbitMQ：强大的开源消息代理，实现了AMQP协议
2. 消息队列：应用程序间异步通信的机制
3. 交换器和路由：消息分发的核心机制
4. 消息确认：保证可靠性的关键机制

概念关系回顾

1. 移动应用与RabbitMQ：移动应用作为生产者发布消息到交换器
2. RabbitMQ与大数据服务：RabbitMQ将消息路由到不同的队列供消费者处理
3. 多种交换器类型：提供灵活的消息路由能力
4. 确认机制与可靠性：确保消息不丢失的关键设计

关键收获

• 理解了RabbitMQ在大数据与移动应用集成中的核心作用
• 掌握了使用RabbitMQ实现松耦合架构的方法
• 学会了如何设计可靠的消息处理系统
• 了解了性能优化和监控的关键技术

思考题：动动小脑筋

思考题一：

假设你正在设计一个社交媒体应用的推送通知系统，如何利用RabbitMQ实现以下功能：

1. 实时推送新消息通知
2. 支持百万级用户在线
3. 处理不同优先级的通知(如私信优先于点赞)
4. 在用户离线时缓存通知

思考题二：

考虑一个移动电商应用在"双十一"期间的场景：

1. 如何设计RabbitMQ架构应对流量激增？
2. 如何处理库存超卖问题？
3. 如何确保订单处理的顺序性？
4. 如何快速定位和处理积压的消息？

思考题三：

在大数据分析场景中：

1. 如何设计RabbitMQ队列来实现实时分析和批量分析的协同工作？
2. 如何处理历史数据的重放？
3. 如何保证数据分析的精确一次(exactly-once)语义？
4. 如何监控消息处理延迟？

附录：常见问题与解答

Q1: RabbitMQ和Kafka有什么区别？什么时候该用哪个？

A1: RabbitMQ更适合：

• 复杂的路由需求
• 低延迟消息传递
• 轻量级部署

Kafka更适合：

• 高吞吐量事件流
• 长期存储和重放
• 流处理场景

两者也可以结合使用，RabbitMQ用于操作消息，Kafka用于事件流。

Q2: 如何确保消息不丢失？

A2: 多重保障机制：

1. 消息持久化(delivery_mode=2)
2. 队列持久化(durable=True)
3. 发布者确认(publisher confirms)
4. 消费者确认(manual acknowledgments)
5. 镜像队列(HA policy)

Q3: RabbitMQ的性能瓶颈通常在哪里？

A3: 常见瓶颈点：

1. 磁盘I/O(持久化消息时)
2. 网络带宽
3. 单个队列的消费者数量
4. 消息大小(建议小于1MB)

Q4: 如何处理积压的消息？

A4: 几种策略：

1. 增加更多消费者
2. 使用惰性队列(lazy queues)减少内存压力
3. 设置TTL自动过期旧消息
4. 将积压消息转移到死信队列单独处理

Q5: 移动应用如何优化与RabbitMQ的交互？

A5: 移动端最佳实践：

1. 使用MQTT协议(更轻量)
2. 实现消息压缩
3. 批量发送消息
4. 智能重试策略
5. 离线消息缓存

扩展阅读 & 参考资料

1. 官方文档：

   • RabbitMQ官方文档: https://www.rabbitmq.com/documentation.html
   • AMQP协议规范: https://www.amqp.org/

2. 书籍：

   • “RabbitMQ in Action” by Alvaro Videla & Jason J. W. Williams
   • “Designing Data-Intensive Applications” by Martin Kleppmann

3. 开源项目：

   • RabbitMQ集群管理工具: https://github.com/rabbitmq/rabbitmq-clusterer
   • RabbitMQ Prometheus exporter: https://github.com/kbudde/rabbitmq_exporter

4. 技术博客：

   • RabbitMQ最佳实践: https://www.cloudamqp.com/blog/part1-rabbitmq-best-practice.html
   • 大规模RabbitMQ部署: https://engineering.riotgames.com/news/going-critical-messaging-rabbitmq

5. 相关技术：

   • MQTT协议: https://mqtt.org/
   • Kafka与RabbitMQ比较: https://www.confluent.io/blog/apache-kafka-vs-rabbitmq-which-one-you-should-use/