引言

在做MQ技术选型的时候，Kafka和RocketMQ是常用的两个消息队列中间件，今天就从架构设计、性能分析、使用场景来比较一下两者的区别，到底该使用哪个MQ？

Kafka最初由LinkedIn开发，后来成为Apache的一个顶级项目，它设计之初就是为处理大规模数据而生，特别擅长于高吞吐量的场景。Kafka广泛应用于日志收集、流式处理、事件驱动架构等多种场景，被许多知名企业采用，如Netflix、Uber和Twitter等。

RocketMQ原为阿里巴巴的内部消息中间件，后来同样成为了Apache的顶级项目。它在保证消息的高可靠性和顺序性方面表现出色，非常适合金融行业等对数据一致性和可靠性要求极高的场景。除此之外，RocketMQ也支持多种消息传递模式，包括顺序消息、延时消息和批量消息，能够满足复杂应用场景的需求。

架构设计

Kafka的基础架构

Apache Kafka 是一个分布式发布-订阅消息系统，设计之初的目标是处理大规模的数据流，并且以高吞吐量和低延迟为特点。Kafka 的架构主要由以下几个部分组成：

1. Producer（生产者）：生产者是发送消息到Broker集群。生产者将消息发送到指定的主题，Kafka根据配置的分区策略（如轮询、按键哈希等）将消息分配到不同的分区。
2. Consumer（消费者）：消费者从Broker读取消息。消费者可以独立运行或分组在一起运行。分组中的消费者共享订阅的主题，Kafka平衡各个消费者的负载，确保每个分区只被组内的一个消费者读取。
3. Broker（消息代理服务器）：Broker是Kafka集群中的一个服务器，负责存储数据和处理对数据的读写请求。每个Broker可以存储一个或多个主题的数据。一个Kafka集群可以包含多个Broker，以提高容量和提供容错能力。
4. Topic（主题）： Topic是逻辑概念。生产者写入消息到指定的Topic，消费者从Topic读取消息。Topic在逻辑上被分割为一个或多个分区，这允许数据在多个Broker之间进行负载均衡。
5. Partition（分区）： 分区是Topic的物理分段，每个分区是一个有序的、不可变的消息日志。分区可以分布在集群中的不同Broker上。每个分区都由一系列有序的、不断增加的消息组成，每条消息都被分配一个顺序的标识符称为偏移量。
6. ZooKeeper：Kafka使用ZooKeeper来维护集群状态、配置信息和进行领导者选举。

RocketMQ的基础架构

RocketMQ主要由四个基本组件构成：

1. NameServer（命名服务器）：NameServer是RocketMQ网络中的注册中心和路由中心，提供轻量级服务发现和路由功能。每个Broker启动时都会在所有NameServer上注册自己的路由信息，包括当前Broker的IP地址、提供的Topic等信息。消费者和生产者通过查询NameServer来获取Topic的路由信息。
2. Broker（消息代理服务器）：Broker是消息处理的核心节点，负责存储消息、验证和服务消息传输。RocketMQ支持多个Broker配置，可以是同步或异步复制数据以确保高可用性。Broker处理大量的数据写入操作，并支持消息的顺序和并行处理。
3. Producer（生产者）：生产者负责发布消息到指定的Topic。RocketMQ支持多种消息发送模式，包括同步发送、异步发送和单向发送（不等待服务器响应）。
4. Consumer（消费者）：消费者从Broker订阅消息并处理它们。RocketMQ支持集群消费和广播消费两种模式。在集群模式下，同一个Consumer Group中的不同Consumer实例平均分摊消息，而在广播模式下，每个Consumer实例都会接收到所有的消息。
5. Topic和Queue： Topic是消息的分类，每个Topic可以分为若干个Queue。RocketMQ通过增加Queue数量来水平扩展Topic的处理能力。

RocketMQ 支持多种消息模式，包括顺序消息、定时/延时消息和批量消息等。此外，RocketMQ 提供了丰富的消息过滤功能，消费者可以根据Tag或者SQL92标准进行消息过滤，极大地增加了其灵活性和应用场景。

Kafka与RocketMQ在设计哲学和优化点上有所不同。Kafka更注重于处理高吞吐量的数据流，而RocketMQ则提供了更为丰富的消息模式和高级功能，特别适合需要高可靠性和复杂消息处理场景的业务。

消息存储机制

Kafka的日志存储机制

• 日志文件： Kafka 所有的消息以日志的形式存储在磁盘上，并且每个Partition都是一个连续的日志文件。
• 追加写入： Kafka采用追加写入的方式存储消息到日志文件中，新消息被添加到文件的末尾，这种方式对于磁盘I/O是非常高效的，因为它大部分是顺序写入，从而极大地提高了写入速度。但是当Partition数量过多时，顺序写就变成了随机写，性能下降。
• 索引文件： 为了快速查找和读取特定消息，Kafka为每个日志文件维护一个索引文件。索引文件存储消息在日志文件中的偏移量和其对应在文件中的物理位置，这样可以在不读取整个日志文件的情况下直接跳转到特定的消息。

RocketMQ的存储设计

RocketMQ的存储系统主要由以下几部分构成：

CommitLog

• 统一存储： 所有Topic的消息都存储在一个名为CommitLog的文件中，每个消息都有一个全局唯一的偏移量。这种设计简化了消息存储的管理，但也要求高效的索引机制来支持快速消息查找。
• 顺序写入： 与Kafka类似，RocketMQ的CommitLog也采用顺序写入的方式，以提高写入效率和减少磁盘I/O操作。顺序写入能显著提高消息存储的性能。
• **定期切割： **RocketMQ定期切分CommitLog和消费队列文件，新的消息写入到新文件中。老旧文件在满足一定条件后可以删除或者归档，以释放存储空间。
• 刷盘策略： RocketMQ提供了同步刷盘和异步刷盘两种策略，用户可以根据业务需求和对性能的要求选择合适的刷盘方式。

消费队列（Consume Queue）

• 索引机制：为了快速检索到CommitLog中的消息，RocketMQ为每个队列（Queue）维护一个消费队列（Consume Queue）。消费队列存储了消息在CommitLog中的偏移量、消息长度和消息标签的哈希码等信息。
• 轻量级设计：消费队列相比于CommitLog要小很多，因为它仅仅存储索引信息，这使得加载和查找效率更高。

索引文件（Index File）

• 可选的索引服务：RocketMQ提供了一个独立的索引服务，用于快速检索具有特定键（如ID、Key或是业务属性）的消息。索引文件存储了键到消息物理位置的映射。
• 快速查询：索引文件加速了基于键的消息查询操作，使得RocketMQ能在大数据量中快速定位消息。

文件回收与存储清理 RocketMQ通过定期清理旧的CommitLog文件和消费队列文件来回收磁盘空间，这些操作基于消息的存储时间和消费状态。

• 定期删除： 系统根据配置的文件保留策略（如时间间隔、文件大小）自动删除旧文件。
• 数据压缩： 在必要时，RocketMQ可以对存储的数据进行压缩，以节省存储空间。

高可用设计

Kafka高可用

• 副本机制： Kafka通过副本（replicas）机制确保数据的安全性。每个Topic可以被配置为一个或多个分区（partitions），每个分区可以有一个或多个副本。副本分布在不同的Broker上，这样即使一个或多个Broker发生故障，Topic的数据也不会丢失。
• 领导者和追随者： 每个分区有一个领导者（leader）和多个追随者（followers）。所有的读写请求都由领导者处理，而追随者则从领导者那里复制数据。如果领导者发生故障，系统会从追随者中选举出新的领导者。
• 控制器（Controller）： 控制器是一个特殊的Broker节点，负责维护领导者的选举和副本状态的管理。如果控制器出现故障，集群中的其他Broker将通过选举产生新的控制器。
• ZooKeeper协调： Kafka使用ZooKeeper来管理集群元数据和进行Broker之间的协调，包括领导者选举和集群成员管理。
• 高水位标记（high watermark）： Kafka为每个分区维护一个“高水位”（high watermark）标记，这是所有同步副本已确认写入的最小偏移量。只有高于高水位的消息才被认为是“提交”的，消费者只能读取到这些已提交的消息。这保证了即使在发生故障的情况下，消费者也不会读取到可能因故障而回滚的消息。

RocketMQ高可用

• 主从架构： 在Broker级别，RocketMQ采用主从架构，其中主Broker负责处理读写请求，而从Broker则负责复制主Broker的数据。如果主Broker宕机，从Broker可以迅速升级为新的主Broker，接管消息服务。
• NameServer的高可用： RocketMQ使用NameServer管理元数据和路由信息，NameServer采用了无状态设计，之间互不备份，每个NameServer独立提供服务。即使部分NameServer出现故障，其他NameServer仍能继续提供服务。
• 同步复制与异步复制： RocketMQ支持同步和异步两种数据复制方式。在同步复制模式下，生产者发送的消息必须被存储在所有同步副本中Broker确认后才返回成功响应，确保了数据的强一致性。异步复制则强调高吞吐量，牺牲了一部分数据安全性。

主要区别：

• 元数据管理：

Kafka强依赖Zookeeper进行集群管理和元数据存储，而RocketMQ则依赖轻量级的NameServer进行路由信息管理，不涉及集群状态管理。

• 数据复制与故障恢复：

Kafka是基于分区，侧重于通过领导者和追随者模式实现数据复制，依赖自动的领导者选举来恢复服务。RocketMQ是基于Broker，提供了主从同步或异步复制，通常需要更多手动干预来切换Master。

• 架构和扩展性：

Kafka的架构设计为分布式系统带来了强大的水平扩展性，适合处理大规模数据流。RocketMQ通过主从复制机制提供高可靠性，适用于交易等对数据一致性要求极高的场景。

消息可靠性保证

Kafka消息可靠性

• 复制（Replication）

Kafka通过在多个Broker中复制每个Topic的Partition来增加数据的可靠性和系统的容错性。这意味着每个Partition都有一个Leader和多个Follower。所有的写操作都通过Leader进行，而Follower从Leader同步数据。如果Leader失败，一个Follower将被自动选举为新的Leader，确保服务的连续性和数据的可用性。

• 确认机制（Acknowledgments）

生产者在发送消息时可以指定不同级别的确认机制来保证消息的可靠传递：

acks=0：生产者在写入消息后不会等待任何服务器的确认，这种模式下消息可能会丢失，但延迟最低。 acks=1：生产者会等待Leader确认消息已被写入本地日志后才考虑完成请求。这种模式下，如果在Follower复制之前Leader发生故障，消息可能会丢失。 acks=all 或 acks=-1：生产者会等待所有同步副本都确认消息已被接收，才认为消息发送成功。这提供了最高的数据可靠性保证。

• 事务支持

从0.11版本开始，Kafka引入了事务API，支持跨多个Partition的原子写操作。这意味着生产者可以发送一批消息，这些消息要么全部成功写入，要么全部失败，从而防止了在处理复杂业务逻辑时出现部分更新的情况。

• 持久化

Kafka默认将所有消息持久化到磁盘，这不仅确保了数据在系统重启后的可恢复性，还能保护数据不受系统故障的影响。Kafka通过顺序写磁盘的方式优化了I/O性能，即使是在高负载下也能保持高吞吐量。

• 高水位（High Watermark）

Kafka为每个Partition维护一个高水位标记，这表示所有同步副本都确认接收到的最小偏移量。消费者只能读取到高水位之前的消息，这保证了消费者只看到已经被所有同步副本确认的消息，增加了读操作的一致性。

RocketMQ消息可靠性

• 复制

同步双写： 在默认设置下，消息被同时写入到主Broker（Master）和从Broker（Slave）。只有当主从Broker都成功写入消息后，生产者才会收到一个成功的响应。这确保了即使主Broker发生故障，消息也不会丢失，因为它已经存在于至少一个从Broker中。

异步复制： 除了同步双写，RocketMQ还支持异步复制模式，这在提高性能和吞吐量方面更为有效，尤其是在对延迟要求不是特别严格的场景下。

• 确认机制（Acknowledgments）

消息确认： RocketMQ 支持端到端的消息确认机制。生产者在发送消息后会等待Broker的确认，只有收到确认后才认为消息发送成功。

重试机制： 如果消息在传输过程中失败（例如，因为网络问题或Broker处理能力达到瓶颈），RocketMQ提供了自动重试机制。生产者可以配置消息的重试次数和重试间隔，以增加消息传递成功的概率。

• 事务消息

RocketMQ 支持事务消息，采用半消息机制（half message），允许生产者在发送消息的同时执行本地事务，然后根据本地事务执行的结果来提交或回滚消息。

延迟消息

• Kafka不支持延迟消息
• RocketMQ提供了多个预定义级别的延迟消息。

可选级别有：

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

示例代码：

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

public class DelayedMessageProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("example_producer_group");
        // 设置NameServer地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        // 创建消息实例，指定Topic、Tag和消息体
        Message message = new Message("TopicTest", "TagA", "Hello RocketMQ, I will be sent after 30 seconds".getBytes());
        // 设置延迟级别为4，即30秒
        message.setDelayTimeLevel(4);

        // 发送消息
        producer.send(message);

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

顺序消息

• Kafka支持单分区有序
• RocketMQ 提供两种类型的顺序消息：
  • 全局顺序消息： 全局顺序消息确保消息全局严格按照发送顺序被消费。实现方式是将所有消息路由到同一个队列（Queue）中。
  • 分区顺序消息： 分区顺序消息确保同一分区内的消息严格按照发送顺序被消费。每个主题可以有多个队列，每个队列保证队列内消息的顺序性。

消息过滤

• Kafka不支持在Broker层面进行消息过滤
• RocketMQ在Broker层面提供了两种消息过滤机制，分别是标签过滤和SQL表达式过滤。

1. 标签过滤（Tag Filtering）

这是RocketMQ最基本也是最常用的消息过滤方式。生产者在发送消息时可以指定一个标签（Tag），消费者在订阅消息时可以指定感兴趣的标签，Broker仅将符合标签的消息推送给消费者。 发送消息时指定标签：

Message msg = new Message("TopicTest", "TagA", "OrderID001", "Hello world".getBytes());

消费者订阅指定标签：

consumer.subscribe("TopicTest", "TagA || TagB");

2. SQL表达式过滤（SQL92 Filtering）

RocketMQ 4.4.0 版本以上支持基于 SQL92 的表达式进行消息过滤，这提供了更为强大和灵活的消息选择能力。生产者在发送消息时可以设置消息属性，消费者可以通过 SQL92 表达式对这些属性进行筛选。 发送消息时设置属性：

Message msg = new Message("TopicTest", "TagA", "OrderID002", "Hello world".getBytes());
msg.putUserProperty("a", String.valueOf(10));

消费者使用 SQL 表达式订阅：

consumer.subscribe("TopicTest", MessageSelector.bySql("a > 5 AND b <> 'abc'"));

对于使用SQL表达式过滤，RocketMQ需要配置Broker启用此功能。在broker.conf中设置：

enablePropertyFilter=true

消息重试

Kafka消息重试机制

Kafka支持生产者发送消息失败的时候自动重试，不支持消费者消费消息失败时重试。

生产者重试配置：

• retries：

这个参数设置了生产者在发送消息时可以重试的次数。默认值通常是 0，表示不进行重试。如果设置为大于0的值，生产者将在发送失败后尝试重新发送消息指定的次数。

• retry.backoff.ms：

这个参数用来设置每次重试之间的时间间隔（以毫秒为单位）。这可以避免在出现暂时性问题时过于频繁地重试，给系统带来不必要的负担。默认值通常是 100ms。

• max.in.flight.requests.per.connection：

此参数定义了生产者在收到服务器响应之前可以发送的最大请求数。如果设置为1，这将保证消息是按照发送的顺序写入服务器的，即使进行了重试。如果大于1，则在高吞吐量的情况下可以提高性能，但可能会导致重试后消息顺序的改变。

生产者重试代码示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class ProducerDemo {
    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.RETRIES_CONFIG, 3);  // 启用重试，重试3次
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);  // 设置重试的时间间隔为300ms
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);  // 保持消息顺序

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key", "value");

        producer.send(record, (RecordMetadata metadata, Exception exception) -> {
            if (exception != null) {
                exception.printStackTrace();  // 处理发送异常
            } else {
                System.out.println("Message sent successfully: " + metadata);
            }
        });

        producer.close();
    }
}

注意事项

• 幂等性：从Kafka 0.11版本开始，支持幂等性生产者，通过配置enable.idempotence=true，可以保证即使进行重试，消息也不会在Kafka中重复。
• 错误处理：应用程序应适当处理重试后仍然失败的情况，比如记录日志、发送告警等。

RocketMQ消息重试机制

RocketMQ 既支持生产者发送消息失败的时候自动重试，也支持消费者消费消息失败时重试。 RocketMQ 生产者发送失败重试机制： 默认重试策略

• 重试次数： RocketMQ 生产者默认会在消息发送失败时自动重试，通常默认重试次数为 2 次（总共发送 3 次：首次发送加上两次重试）。
• 重试间隔： 默认的重试间隔时间是 3000 毫秒（3 秒），即在初次发送失败后，会在 3 秒后进行第一次重试。

可以根据实际需要配置生产者的重试次数和重试间隔。这通常在创建生产者实例时进行设置：

public static void main(String[] args) throws MQClientException {
    // 创建消息生产者，指定生产者组名
    DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
    // 设置 NameServer 地址
    producer.setNamesrvAddr("localhost:9876");
    
    // 设置重试次数，默认2次，设置为5次
    producer.setRetryTimesWhenSendFailed(5);
    
    // 设置消息发送超时时间，超过这个时间未发送成功则不再重试，默认为3000ms
    producer.setSendMsgTimeout(4000);
    
    // 启动生产者
    producer.start();

    // 创建消息
    Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET));
    
    // 发送消息
    try {
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
    } catch (Exception e) {
        e.printStackTrace();
    }

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

RocketMQ 消费者消费失败重试机制：

当消费者从队列中拉取到消息并且在处理过程中失败（通常是业务逻辑抛出异常或者返回了错误状态），RocketMQ 提供了自动的消息重试机制。这意味着消息不会被立即标记为“已消费”，而是会重新被放入队列，稍后再次投递给消费者。

• 重试延迟：

RocketMQ 为消费失败的消息设置了一系列递增的延迟时间等级，例如，首次重试可能延迟 10 秒，随后 30 秒、1 分钟、2 分钟等。 默认情况下，RocketMQ 提供了 16 级延迟时间，最长可以延迟两个小时。

• 重试次数：

如果消息连续多次重试仍然失败，当重试次数达到上限后（默认是 16 次），消息将不再进入重试队列。 这些“死信消息”会被转移到一个特殊的死信队列（DLQ，Dead-Letter Queue），开发者可以对这些消息进行特殊处理。 非顺序消息重试间隔如下：

消费者重试代码示例：

public class ConsumerDemo {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    for (MessageExt msg : msgs) {
                        String body = new String(msg.getBody(), "UTF-8");
                        System.out.println("Receive message: " + body);
                        // 业务逻辑处理
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();
                    // 告诉MQ这条消息处理失败，需要稍后重新消费
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });

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

消息回溯

消息回溯指的是消费者重新消费已经消费过的消息。这个功能对于处理消费失败、消息处理错误或者需要重新处理数据的场景非常有用。

Kafka和RocketMQ都支持以下两种消息回溯：

• 基于偏移量回溯：

消费者可以直接指定队列的偏移量来回溯消息。这种方式需要消费者知道具体的偏移量。

• 基于时间戳回溯：

消费者根据时间戳来重置消费进度。这种方式适用于希望从某一特定时间点重新开始消费消息的场景。

事务消息

Kafka事务消息

Kafka 0.11 版本开始支持事务消息，允许生产者在单个或多个分区中原子地写入消息。通过事务消息，生产者可以确保消息要么全被发送，要么全不发送，从而避免了在失败时消息部分发送的问题。

Kafka 事务消息的关键概念

• 事务ID：每个事务生产者都被分配一个唯一的事务ID。这个ID用来标识生产者的事务状态，并保证即使在发生故障后，也能恢复并继续处理事务。
• 事务协调器：Kafka集群中的每个事务生产者都有一个事务协调器（Transaction Coordinator）与之对应。协调器负责管理所有与其事务ID相关的事务状态。
• 生产者幂等性：事务生产者在 Kafka 中自动启用幂等性。幂等性保证了即使生产者发送相同消息多次，消息也只会被写入一次。

如何使用 Kafka 事务消息

• 配置事务生产者：启用事务需要在生产者配置中设置 transactional.id 和开启幂等性。
• 初始化事务：通过调用 initTransactions() 方法初始化事务环境。
• 开始事务：通过调用 beginTransaction() 开始一个新的事务。
• 发送消息：在事务中正常发送消息。
• 提交或中止事务：根据业务逻辑处理结果，调用 commitTransaction() 或 abortTransaction() 来提交或中止事务。

示例代码

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaTransactionalProducerExample {
    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, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "prod-1"); // 指定事务ID
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // 启用幂等性

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // 1. 初始化事务
        producer.initTransactions();

        try {
            // 2. 开启事务
            producer.beginTransaction();
            for (int i = 0; i < 100; i++) {
                // 3. 同时发送到多个分区
                producer.send(new ProducerRecord<>("your-topic", Integer.toString(i), "test message - " + i));
            }
            // 4. 提交事务
            producer.commitTransaction();
        } catch (Exception e) {
            // 5. 中止事务
            producer.abortTransaction();
        } finally {
            producer.close();
        }
    }
}

注意事项

• 事务超时：事务生产者必须在配置的事务超时时间内完成事务，否则 Kafka 会认为事务失败并自动中止它。
• 单一生产者规则：事务ID 应唯一对应单一生产者实例，以避免并发冲突和潜在的数据不一致问题。
• 事务与消费者：确保消费者正确处理事务消息，例如使用 read_committed 配置来只消费已提交的消息。

RocketMQ 事务消息

RocketMQ采用半消息机制，实现了事务消息，就是把本地事务和生产者发送消息放在一个事务中。 RocketMQ 事务消息工作原理

1. 半消息（Half Message）： 事务消息首先被发送为“半消息”，这意味着消息被Broker接收但对消费者不可见。
2. 执行本地事务： 一旦半消息被成功发送，生产者客户端将执行本地事务逻辑（如数据库操作）。
3. 提交或回滚： 根据本地事务执行的结果，生产者决定是提交还是回滚事务消息：
   • 如果本地事务成功，生产者发送提交消息指令给Broker，使得半消息对消费者可见。
   • 如果本地事务失败，生产者发送回滚消息指令给Broker，Broker将删除半消息。
4. 消息状态回查： 如果Broker没有收到最终的提交或回滚指令（可能由于生产者崩溃等原因），Broker将向生产者查询该半消息对应的本地事务状态。

示例代码

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException {
        // 创建事务生产者，指定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("producer_group");
        producer.setNamesrvAddr("localhost:9876");

        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 执行本地事务逻辑
                // 返回事务状态
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 提供事务状态的检查逻辑
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        producer.start();

        // 创建消息
        Message msg = new Message("TopicTest", "TagA", "Key", "Transaction Message".getBytes());

        // 发送事务消息
        TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
        System.out.printf("%s%n", sendResult);

        producer.shutdown();
    }
}

集群扩展

Kafka 集群扩展

1. 统一的 Broker 角色：

• Kafka 的 Broker 同时负责消息存储、处理和路由信息的管理。每个 Broker 都可以处理客户端的请求，存储消息数据，以及处理其他 Broker 的数据复制请求。

1. 复制机制：

• Kafka 采用分区和副本的方式进行数据复制。每个主题被分为多个分区，每个分区可以有一个或多个副本，复制策略可以配置为同步或异步。

1. 扩展性：

• 扩展 Kafka 集群主要是通过增加更多的 Broker 来完成。新的 Broker 加入后，可以通过重新平衡分区来分散数据和请求负载。

1. 依赖于 Zookeeper：

• Kafka 使用 Zookeeper 进行集群管理和协调，虽然最新版本的 Kafka 正在尝试去除对 Zookeeper 的依赖。

RocketMQ 集群扩展

1. 角色区分：

• RocketMQ 明确区分了 Broker 和 NameServer 的角色。Broker 负责消息存储和传输，而 NameServer 提供路由信息和服务发现功能，不参与消息传递。

1. 主从同步：

• RocketMQ 支持主从同步，其中 Master Broker 处理读写请求，Slave Broker 主要用于数据同步和故障恢复。

1. 扩展方式：

• 扩展 RocketMQ 集群通常涉及添加更多的 Broker（Master/Slave）和可能的 NameServer。这种方式有助于提升集群的容错能力和数据的可用性。

1. 灵活的部署：

• 可以灵活地部署多个 NameServer 来提高服务的可用性，但 NameServer 之间不进行数据同步。

核心区别

• 依赖服务：RocketMQ 使用 NameServer 作为独立的路由和服务发现层，而 Kafka 使用 Zookeeper 作为协调服务。
• 数据同步：RocketMQ 的主从架构与 Kafka 的分区副本策略提供了不同的数据同步和故障恢复机制。
• 扩展操作：RocketMQ 在扩展时可能需要同时增加 Broker 和 NameServer，而 Kafka 的扩展更多关注于增加 Broker 和分区重新平衡。

使用场景

RocketMQ 使用场景

1. 事务消息：

• RocketMQ 提供原生支持的事务消息特别适合需要处理复杂业务逻辑的场景，如电子商务中的订单系统，可以在处理业务逻辑失败时进行消息回滚。

1. 顺序消息：

• RocketMQ 支持严格的顺序消息，非常适合需要消息严格顺序消费的场景，如金融行业的交易和支付系统。

1. 广播消息：

• RocketMQ 支持广播消息发送，适用于发送如广告信息、系统通知等到多个接收者的场景。

1. 定时、延迟消息：

• RocketMQ 支持定时或延迟消息传递，适合需要在指定时间执行任务的应用，例如定时推送、预约提醒等。

1. 可靠性和可用性较高的应用：

• RocketMQ 的设计注重高可用性和服务的稳定性，适合银行、股票交易和电信运营商等对消息丢失敏感度极高的行业。

Kafka 使用场景

1. 日志聚合：

• Kafka 常用于日志数据的收集和聚合，适用于需要高吞吐量处理日志文件的场景，如中大型网站的用户活动跟踪、应用日志集中管理等。

1. 流式处理：

• Kafka 与流处理框架（如 Apache Flink、Apache Storm 和 Kafka Streams）结合，提供实时数据流处理能力，适合实时分析和监控系统。

1. 事件驱动架构：

• Kafka 支持高吞吐的事件发布和订阅，适用于构建微服务架构中的事件驱动系统，可以作为各个服务之间解耦的通信中间件。

1. 数据湖或数据仓库的数据集成：

• Kafka 可以作为数据管道，将数据从多个源头实时传输到大数据平台（如 Hadoop 或 Spark），支持大数据分析和数据挖掘。

1. 分布式系统的冗余备份：

• Kafka 的数据复制特性适用于需要在多个地理位置进行冗余备份的系统，以提高数据的可靠性和系统的灾难恢复能力。

区别

选择 RocketMQ 或 Kafka 主要取决于具体的业务需求、系统要求以及团队的技术栈偏好。如果需要处理具有复杂业务逻辑的事务性消息，或需要精确控制消息顺序和定时发送的功能，RocketMQ 可能是更合适的选择。 而如果应用场景更侧重于高吞吐量的数据处理，如日志收集、实时数据流处理或事件驱动的微服务架构，Kafka 则可能是更优的选择。