MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

RocketMQ 与 Kafka 性能对比分析

2024-03-276.2k 阅读

1. 架构设计差异对性能的影响

1.1 RocketMQ 的架构设计

RocketMQ 的架构主要由 NameServer、Broker、Producer 和 Consumer 组成。NameServer 作为轻量级的服务发现和路由中心，集群中的每个 NameServer 都保存全量的路由信息，相互之间无通信，无状态，Broker 启动时向所有 NameServer 注册，并定时上报心跳信息。Producer 和 Consumer 通过 NameServer 获取 Broker 的路由信息，然后直接与 Broker 进行数据交互。

这种架构设计使得 RocketMQ 在集群管理和故障恢复方面具有一定的优势。例如，当某个 NameServer 节点出现故障时，由于其他 NameServer 节点保存了相同的路由信息，Producer 和 Consumer 依然可以正常获取路由，从而保证消息的正常生产和消费，不会影响整体性能。

在存储方面，RocketMQ 的 Broker 采用了 CommitLog 与 ConsumeQueue 分离的存储结构。CommitLog 是所有消息的物理存储文件，顺序写，极大提高了写入性能。而 ConsumeQueue 则是消息消费的逻辑队列，保存了指向 CommitLog 中消息的物理偏移量等信息，这种结构在保证写入性能的同时，也提高了消息的查询和消费效率。

1.2 Kafka 的架构设计

Kafka 的架构包括 Zookeeper、Broker、Producer 和 Consumer。Zookeeper 在 Kafka 中扮演着集群管理、元数据存储等重要角色。Broker 负责消息的存储和转发，Producer 和 Consumer 通过与 Broker 交互来生产和消费消息。

Kafka 的 Broker 采用了分区（Partition）的概念，每个 Topic 可以分为多个分区，分区分布在不同的 Broker 上，以实现负载均衡和高可用性。消息在分区内是有序的，但跨分区无法保证顺序。在存储方面，Kafka 的每个分区对应一个日志文件，消息以追加的方式写入日志文件，同样利用了磁盘顺序写的特性来提高写入性能。

然而，Zookeeper 在 Kafka 架构中是一个单点故障隐患。虽然 Kafka 可以通过增加 Zookeeper 节点来提高可靠性，但当 Zookeeper 集群出现故障时，可能会导致 Kafka 集群的元数据管理出现问题，进而影响消息的生产和消费，对性能产生较大影响。

2. 消息发送性能对比

2.1 同步发送性能

在 RocketMQ 中，同步发送消息是指 Producer 发送消息后，等待 Broker 返回确认响应，才继续发送下一条消息。以下是 RocketMQ 同步发送消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class RocketMQSyncProducer {
    public static void main(String[] args) throws Exception {
        // 创建一个生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("sync_producer_group");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建一条消息
            Message message = new Message("sync_topic", "TagA", ("Hello RocketMQ Sync " + i).getBytes("UTF-8"));
            // 同步发送消息
            SendResult sendResult = producer.send(message);
            System.out.println("SendResult: " + sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

在 Kafka 中，同步发送消息同样是 Producer 发送消息后等待 Broker 的确认。以下是 Kafka 同步发送消息的代码示例：

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class KafkaSyncProducer {
    public static void main(String[] args) {
        // 设置生产者属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 创建生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("sync_topic", "Key" + i, "Hello Kafka Sync " + i);
            try {
                // 同步发送消息
                RecordMetadata metadata = producer.send(record).get();
                System.out.println("Sent message to partition " + metadata.partition() + " with offset " + metadata.offset());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        // 关闭生产者
        producer.close();
    }
}

在同步发送场景下，RocketMQ 和 Kafka 的性能差异主要体现在网络通信和确认机制上。RocketMQ 的 NameServer 架构使得 Producer 与 Broker 的连接相对稳定，并且在消息确认方面，RocketMQ 的 Broker 可以快速返回确认响应。而 Kafka 由于依赖 Zookeeper 进行元数据管理，在某些情况下，例如 Zookeeper 集群负载较高时，可能会导致消息确认延迟，从而影响同步发送的性能。

2.2 异步发送性能

RocketMQ 的异步发送允许 Producer 在发送消息后不等待 Broker 的确认响应，直接继续发送下一条消息，通过回调函数来处理发送结果。以下是 RocketMQ 异步发送消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class RocketMQAsyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("async_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message message = new Message("async_topic", "TagA", ("Hello RocketMQ Async " + i).getBytes("UTF-8"));
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("SendResult: " + sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("Send failed: " + e);
                }
            });
        }

        // 为了确保所有异步发送的消息都被处理，这里设置一个短暂的睡眠时间
        Thread.sleep(1000);
        producer.shutdown();
    }
}

Kafka 也支持异步发送，通过回调函数来处理发送结果。以下是 Kafka 异步发送消息的代码示例：

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class KafkaAsyncProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("async_topic", "Key" + i, "Hello Kafka Async " + i);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    if (e != null) {
                        System.out.println("Send failed: " + e);
                    } else {
                        System.out.println("Sent message to partition " + metadata.partition() + " with offset " + metadata.offset());
                    }
                }
            });
        }

        // 为了确保所有异步发送的消息都被处理，这里设置一个短暂的睡眠时间
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        producer.close();
    }
}

在异步发送场景下，两者都能够充分利用异步特性提高发送性能。然而，RocketMQ 在异步发送时，其内部的线程模型和网络优化机制可以使得消息发送的吞吐量更高。RocketMQ 的 Broker 端对于异步消息的处理也更加高效，能够快速处理并返回结果给回调函数。而 Kafka 在高并发异步发送时，由于分区和副本机制的复杂性，可能会出现一些性能波动。

2.3 单向发送性能

RocketMQ 的单向发送是指 Producer 发送消息后，不关心 Broker 的确认响应，直接继续执行后续操作，主要用于一些对可靠性要求不高但追求极致性能的场景。以下是 RocketMQ 单向发送消息的代码示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class RocketMQOnewayProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("oneway_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message message = new Message("oneway_topic", "TagA", ("Hello RocketMQ Oneway " + i).getBytes("UTF-8"));
            producer.sendOneway(message);
        }

        // 为了确保所有单向发送的消息都被发送出去，这里设置一个短暂的睡眠时间
        Thread.sleep(1000);
        producer.shutdown();
    }
}

Kafka 虽然没有严格意义上的单向发送模式，但可以通过设置 Producer 的 acks = 0 来近似实现单向发送，即 Producer 发送消息后不等待 Broker 的任何确认。以下是类似 Kafka 单向发送消息的代码示例：

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class KafkaOnewayProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.ACKS_CONFIG, "0");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("oneway_topic", "Key" + i, "Hello Kafka Oneway " + i);
            producer.send(record);
        }

        // 为了确保所有单向发送的消息都被发送出去，这里设置一个短暂的睡眠时间
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        producer.close();
    }
}

在单向发送场景下，RocketMQ 的性能优势明显。RocketMQ 针对单向发送进行了专门的优化，其底层网络通信和存储写入机制能够快速处理大量的单向消息，实现极高的吞吐量。而 Kafka 在这种近似单向发送的模式下，由于其架构中分区和副本的管理机制，虽然也能有较高的性能，但相比 RocketMQ 还是稍逊一筹。

3. 消息消费性能对比

3.1 顺序消费性能

RocketMQ 对顺序消费有很好的支持。在 RocketMQ 中，一个 Topic 可以包含多个队列（Queue），Producer 可以通过 MessageQueueSelector 将消息发送到指定的队列，Consumer 可以按照队列顺序消费消息，从而保证消息的顺序性。以下是 RocketMQ 顺序消费的代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketMQOrderlyConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("orderly_topic", "*");

        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("Consume message: " + new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.out.println("RocketMQ Orderly Consumer Started.");
    }
}

Kafka 在单个分区内可以保证消息的顺序性，但跨分区无法保证。要实现类似 RocketMQ 的全局顺序消费，需要将所有消息发送到同一个分区，这在高并发场景下会严重影响性能。以下是 Kafka 在单个分区内顺序消费的代码示例：

import org.apache.kafka.clients.consumer.*;
import java.util.Collections;
import java.util.Properties;

public class KafkaPartitionOrderlyConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "partition_orderly_consumer_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("partition_orderly_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Consume message: " + record.value());
            }
        }
    }
}

在顺序消费性能方面，RocketMQ 由于其设计理念，能够在保证消息顺序的同时，通过合理的队列管理和负载均衡，实现较高的消费性能。而 Kafka 要实现全局顺序消费面临分区瓶颈，在多分区情况下无法保证全局顺序，这限制了其在顺序消费场景下的应用。

3.2 并发消费性能

RocketMQ 支持并发消费，Consumer 可以配置多个线程来并行处理消息。以下是 RocketMQ 并发消费的代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketMQConcurrentConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("concurrent_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("concurrent_topic", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("Consume message: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 设置消费线程数
        consumer.setConsumeThreadMin(10);
        consumer.setConsumeThreadMax(20);

        consumer.start();
        System.out.println("RocketMQ Concurrent Consumer Started.");
    }
}

Kafka 同样支持并发消费，通过多个 Consumer 实例组成 Consumer Group 来并行消费不同分区的消息。以下是 Kafka 并发消费的代码示例：

import org.apache.kafka.clients.consumer.*;
import java.util.Collections;
import java.util.Properties;

public class KafkaConcurrentConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "concurrent_consumer_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("concurrent_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Consume message: " + record.value());
            }
        }
    }
}

在并发消费性能方面，Kafka 由于其分区设计，天然适合高并发消费场景。多个 Consumer 实例可以并行消费不同分区的消息，充分利用集群资源，实现高吞吐量。RocketMQ 虽然也支持并发消费，但在大规模并发场景下，其性能可能略低于 Kafka。这是因为 Kafka 的分区机制使得消息的并行处理更加直接，而 RocketMQ 在处理并发消费时，需要在队列层面进行一些协调和管理。

4. 存储性能对比

4.1 磁盘写入性能

RocketMQ 的 Broker 在存储消息时，采用 CommitLog 顺序写的方式，极大提高了磁盘写入性能。CommitLog 作为所有消息的物理存储文件，顺序记录消息，减少了磁盘 I/O 寻道时间。同时，RocketMQ 采用了 PageCache 机制，将部分数据缓存在内存中，进一步提高了写入性能。

Kafka 的 Broker 同样利用了磁盘顺序写的特性，每个分区对应一个日志文件，消息以追加的方式写入日志文件。Kafka 也依赖操作系统的 PageCache 来提高写入性能。然而，由于 Kafka 的分区机制，在高并发写入时，可能会因为多个分区同时写入而导致磁盘 I/O 竞争，影响整体写入性能。相比之下，RocketMQ 的 CommitLog 统一存储方式在一定程度上可以缓解这种 I/O 竞争问题，使得在大规模写入场景下，RocketMQ 的磁盘写入性能更具优势。

4.2 消息存储结构对查询性能的影响

RocketMQ 的 ConsumeQueue 作为消息消费的逻辑队列，保存了指向 CommitLog 中消息的物理偏移量等信息。这种结构使得在查询消息时，Consumer 可以先从 ConsumeQueue 中快速定位到消息在 CommitLog 中的位置，然后读取消息，提高了查询效率。

Kafka 在查询消息时，由于其分区和日志文件的结构，需要从分区对应的日志文件中按偏移量查找消息。虽然 Kafka 的日志文件结构简单，但在大规模数据存储时，查询特定消息可能需要遍历较长的日志文件，相比 RocketMQ 的 ConsumeQueue 机制，查询性能可能稍逊一筹。

5. 可靠性与可用性对性能的影响

5.1 RocketMQ 的可靠性与可用性

RocketMQ 通过多副本机制来保证消息的可靠性。Broker 集群中的每个 Broker 可以配置多个 Slave 节点，Master 节点负责接收和处理消息，Slave 节点实时从 Master 节点同步数据。当 Master 节点出现故障时，Slave 节点可以自动切换为 Master 节点，保证消息的正常生产和消费，从而实现高可用性。

在消息可靠性方面，RocketMQ 支持同步刷盘和异步刷盘两种方式。同步刷盘保证消息在写入磁盘后才返回确认响应，确保消息不丢失，但会稍微影响性能。异步刷盘则在消息写入 PageCache 后就返回确认响应，提高了写入性能，但存在一定的消息丢失风险。

5.2 Kafka 的可靠性与可用性

Kafka 通过副本机制来保证可靠性和可用性。每个分区可以配置多个副本，其中一个副本为 Leader，其他副本为 Follower。Producer 发送的消息首先写入 Leader 副本，Follower 副本从 Leader 副本同步数据。当 Leader 副本出现故障时，Kafka 会从 Follower 副本中选举出新的 Leader 副本，保证消息的正常生产和消费。

Kafka 的可靠性也依赖于 ACK 机制。Producer 可以通过设置 acks 参数来控制消息的确认级别，例如 acks = all 表示所有副本都确认收到消息后，Producer 才认为消息发送成功，这种方式保证了消息的高可靠性，但在一定程度上会影响性能。

总体而言，RocketMQ 和 Kafka 在可靠性和可用性方面都有完善的机制。然而，这些机制在实现过程中对性能产生了不同程度的影响。RocketMQ 的同步刷盘和 Kafka 的 acks = all 等强可靠性配置都会在一定程度上降低性能，而异步刷盘和较低的 ACK 级别虽然提高了性能，但牺牲了部分可靠性。在实际应用中，需要根据业务需求来平衡可靠性和性能之间的关系。

6. 应用场景与性能适配

6.1 对顺序性要求高的场景

如金融交易、订单处理等场景，对消息的顺序性要求极高。在这种场景下，RocketMQ 由于其良好的顺序消费支持，能够保证消息的严格顺序，且在顺序消费性能方面表现出色，因此更适合此类场景。

6.2 高并发、大数据量的场景

对于日志收集、实时数据处理等高并发、大数据量的场景，Kafka 的分区机制和高并发消费性能使其能够充分利用集群资源，实现高吞吐量。虽然 RocketMQ 也能处理高并发场景，但在大规模数据量和高并发写入时，Kafka 的性能优势更为明显。

6.3 对可靠性和低延迟要求并存的场景

在一些对消息可靠性要求高，同时又希望有较低延迟的场景中，RocketMQ 可以通过合理配置同步刷盘和异步刷盘，以及优化网络通信等方式，在保证一定可靠性的同时，尽量降低延迟。而 Kafka 可以通过调整 ACK 机制和副本配置，在可靠性和延迟之间找到平衡。但总体来说，RocketMQ 在这种场景下，由于其架构设计和存储机制，能够更好地满足可靠性和低延迟的双重要求。