Java的RocketMQ使用

🕗 发布于 2024-10-20 00:34 java-rocketmq java rocketmq

在 Spring Boot 中，RocketMQ 和 Kafka 都是常用的消息中间件，它们的使用方法有一些相似之处，也有各自的特点。

一、RocketMQ 在 Spring Boot 中的使用

引入依赖

在项目的pom.xml文件中添加 RocketMQ 的依赖。

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

配置 RocketMQ
- 在application.properties或application.yml文件中配置 RocketMQ 的相关参数，如 namesrvAddr（NameServer 地址）等。
```
rocketmq.name-server=127.0.0.1:9876
```

生产者

创建一个生产者类，使用@Resource注入RocketMQTemplate。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RocketMQProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String message) {
        rocketMQTemplate.convertAndSend(topic, message);
    }
}

消费者

创建一个消费者类，使用@RocketMQMessageListener注解指定监听的主题和消费组。

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "your_topic", consumerGroup = "your_consumer_group")
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        // 处理接收到的消息
        System.out.println("Received message: " + message);
    }
}

二、Kafka 在 Spring Boot 中的使用

引入依赖

在pom.xml文件中添加 Kafka 的依赖。

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.12</version>
</dependency>

配置 Kafka
- 在application.properties或application.yml文件中配置 Kafka 的相关参数，如 bootstrapServers（Kafka 服务器地址）等。
```
spring.kafka.bootstrap-servers=127.0.0.1:9092
```

生产者

创建一个生产者类，使用@Resource注入KafkaTemplate。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

消费者

创建一个消费者类，使用@KafkaListener注解指定监听的主题和消费组。

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {
    @KafkaListener(topics = "your_topic", groupId = "your_consumer_group")
    public void onMessage(String message) {
        // 处理接收到的消息
        System.out.println("Received message: " + message);
    }
}

总的来说，RocketMQ 和 Kafka 在 Spring Boot 中的使用都比较方便，具体选择哪种消息中间件可以根据项目的实际需求来决定。RocketMQ 在一些场景下可能具有高吞吐量、低延迟等优势，而 Kafka 则在大规模分布式系统中被广泛应用，具有高可靠性和可扩展性。

二、如何保证消息队列顺序性

1、发送端保证顺序性

合理设计业务
- 确保具有顺序性要求的消息被发送到同一个主题（Topic）的同一个队列（Queue）中。比如，将同一类业务的消息按照特定规则进行分类，使得它们都进入相同的队列。
- 一个业务场景的消息尽量由一个发送端来发送消息，避免多个发送端发送可能导致的乱序。
使用同步发送
- 在发送消息时，使用同步发送方式send(Message msg, long timeout)，确保消息成功发送后再进行下一个消息的发送。这样可以避免异步发送可能导致的消息乱序情况。

2、消费端保证顺序性

单线程消费

消费者在消费消息时，采用单线程的方式进行消费。这样可以确保同一队列中的消息按照发送的顺序被依次处理。

@Component
@RocketMQMessageListener(topic = "your_topic", consumerGroup = "your_consumer_group")
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        // 处理接收到的消息
        System.out.println("Received message: " + message);
    }
}

在实际应用中，可以将消费逻辑放在一个单独的方法中，然后在这个方法中进行顺序处理，确保消息的顺序性。

避免并发处理
- 确保在消费消息的过程中，不会出现并发处理的情况。比如，不要在消费消息的同时启动其他异步任务或者多线程处理，以免破坏消息的顺序性。

3、设置队列数量

控制队列数量
- 如果业务对消息顺序性要求非常严格，可以考虑减少主题下的队列数量。通常情况下，一个主题可以包含多个队列，消息会被随机分发到不同的队列中。如果队列数量较少，那么消息更有可能被发送到同一个队列中，从而更容易保证顺序性。

通过以上方法，可以在一定程度上保证 RocketMQ 消息的顺序性。但需要注意的是，保证消息顺序性可能会牺牲一定的性能和吞吐量，因此需要根据实际业务需求进行权衡和选择。

一、如何确保消息队列的可靠性

1、发送端

同步发送与确认

使用同步发送方式send(Message msg, long timeout)，该方法会等待消息发送成功的确认，确保消息被正确地发送到 Broker。如果发送失败或超时，可以进行重试或其他错误处理操作。

try {
    SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
    System.out.println("Message sent successfully: " + sendResult);
} catch (Exception e) {
    System.out.println("Failed to send message: " + e.getMessage());
    // 进行重试或其他错误处理
}

事务消息

对于一些需要保证事务一致性的场景，可以使用 RocketMQ 的事务消息机制。发送事务消息分为两个阶段，首先发送半事务消息，然后执行本地事务，根据本地事务的结果决定提交或回滚事务消息。

@Service
public class TransactionProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendTransactionMessage() {
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("transactionTopic", new Message<>("transactionMessage"), null);
        System.out.println("Transaction message sent: " + result);
    }
}

2、Broker 端

持久化存储
- RocketMQ 支持消息的持久化存储，可以将消息存储在磁盘上，以防止消息丢失。通过配置broker.conf文件中的flushDiskType参数，可以选择同步刷盘或异步刷盘方式。同步刷盘可以保证消息在写入磁盘后才返回成功响应，但会影响性能；异步刷盘可以提高性能，但在系统故障时可能会丢失部分未刷盘的消息。
高可用部署
- 部署多主多从的 RocketMQ 集群，当主节点出现故障时，从节点可以自动切换为主节点，保证消息服务的可用性。同时，可以配置主从同步方式，确保消息在主从节点之间的可靠同步。

3、消费端

消费确认

消费者在成功处理消息后，需要向 Broker 发送消费确认。可以通过设置consumeMode为CONSUME_PASSIVELY（被动消费模式），并在处理完消息后手动调用acknowledge()方法进行确认。如果消费失败，可以选择重试或者将消息发送到死信队列进行后续处理。

@Component
@RocketMQMessageListener(topic = "your_topic", consumerGroup = "your_consumer_group")
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        try {
            // 处理消息
            System.out.println("Received message: " + message);
            // 确认消费成功
            getRocketMQListenerContainer().acknowledge();
        } catch (Exception e) {
            System.out.println("Failed to process message: " + e.getMessage());
            // 可以选择重试或者发送到死信队列
        }
    }
}

重试机制
- 配置消费者的重试次数和重试时间间隔，当消费失败时，RocketMQ 会自动进行重试。可以在application.properties或application.yml中配置rocketmq.retry.times和rocketmq.retry.interval参数来控制重试策略。

通过以上措施，可以在不同阶段保证 RocketMQ 消息的可靠性，确保消息在生产、存储和消费过程中不会丢失或出现错误。

三、保证消息处理的幂等性
在 RocketMQ 中，可以通过以下几种方式来保证消息处理的幂等性：

1、业务层面设计

使用唯一标识

在业务中为每条消息生成一个唯一的标识，比如使用业务流水号、订单号等作为消息的唯一标识。在消费消息时，先根据这个唯一标识判断该消息是否已经被处理过。如果已经处理过，则直接忽略该消息。
例如在电商系统中，订单创建的消息可以使用订单号作为唯一标识。消费者在处理消息时，先查询数据库中是否存在该订单号对应的处理记录，如果存在则说明该消息已经被处理过，不再重复处理。

@Service
public class OrderProcessingService {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    public void processOrderMessage(String orderId) {
        boolean isProcessed = isOrderProcessed(orderId);
        if (isProcessed) {
            return;
        }
        // 处理订单逻辑
        System.out.println("Processing order: " + orderId);
        markOrderAsProcessed(orderId);
    }

    private boolean isOrderProcessed(String orderId) {
        int count = jdbcTemplate.queryForObject(
                "SELECT COUNT(*) FROM processed_orders WHERE order_id =?",
                Integer.class, orderId);
        return count > 0;
    }

    private void markOrderAsProcessed(String orderId) {
        jdbcTemplate.update(
                "INSERT INTO processed_orders (order_id) VALUES (?)",
                orderId);
    }
}

利用数据库约束

可以在数据库中使用唯一索引、主键约束等方式来保证业务数据的唯一性。在处理消息时，如果违反了这些约束，则说明该消息已经被处理过，不再重复处理。
比如在用户注册的场景中，可以在数据库的用户表中使用用户名或邮箱作为唯一索引。当消费用户注册的消息时，尝试插入用户数据，如果插入失败（因为违反唯一索引约束），则说明该用户已经注册过，不再重复处理。

@Service
public class UserRegistrationService {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    public void registerUser(String username, String password) {
        try {
            jdbcTemplate.update(
                    "INSERT INTO users (username, password) VALUES (?,?)",
                    username, password);
        } catch (DataIntegrityViolationException e) {
            // 处理插入失败的情况，可能是用户已存在
            System.out.println("User already exists: " + username);
        }
    }
}

2、技术层面实现

分布式锁

可以使用分布式锁来保证同一时间只有一个消费者实例在处理特定的消息。在处理消息之前，先获取分布式锁，如果获取成功则处理消息，处理完成后释放锁。如果获取锁失败，则说明该消息正在被其他实例处理，当前实例可以选择等待或者直接忽略该消息。
可以使用 Redis 或 Zookeeper 等实现分布式锁。以 Redis 为例，可以使用 SETNX 命令来实现分布式锁。

@Service
public class MessageProcessingService {
    @Autowired
    private StringRedisTemplate redisTemplate;

    public void processMessage(String messageId) {
        String lockKey = "message_lock_" + messageId;
        boolean locked = tryLock(lockKey);
        if (!locked) {
            return;
        }
        try {
            boolean isProcessed = isMessageProcessed(messageId);
            if (isProcessed) {
                return;
            }
            // 处理消息逻辑
            System.out.println("Processing message: " + messageId);
            markMessageAsProcessed(messageId);
        } finally {
            releaseLock(lockKey);
        }
    }

    private boolean tryLock(String key) {
        return redisTemplate.opsForValue().setIfAbsent(key, "locked", Duration.ofSeconds(30));
    }

    private void releaseLock(String key) {
        redisTemplate.delete(key);
    }

    private boolean isMessageProcessed(String messageId) {
        // 判断消息是否已处理的逻辑
        return false;
    }

    private void markMessageAsProcessed(String messageId) {
        // 标记消息已处理的逻辑
    }
}

通过以上方法，可以有效地保证 RocketMQ 消息处理的幂等性，避免因重复消费消息而导致的业务数据不一致问题。

原文地址：https://blog.csdn.net/mozf881/article/details/143032154

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：SpringCloud无介绍快使用，单机Eureka服务注册中心cloud-eureka-server7001搭建（十）
下一篇：perl双引号内字符串的反斜线转义

leetcode289:生命游戏
根据，简称为，是英国数学家约翰·何顿·康威在 1970 年发明的细胞自动机。给定一个包含m × n个格子的面板，每一个格子都可以看成是一个细胞。每个细胞都具有一个初始状态：1即为（live），或0即为
阅读更多2024-10-20
MongoDB数据恢复
注意：两个MongoDB的版本要一致，本文使用的是mongo:4.2.24。先把K8S上面的MongoDB 容器停止（可以把副本改成0）。1、将容器挂载MongoDB的数据目录备份到本地。经常是数据文
阅读更多2024-10-20
C#中实现事务
C#中实现事务
阅读更多2024-10-20
【LeetCode每日一题】——560.和为 K 的子数组
给你一个整数数组 nums 和一个整数 k ，请你统计并返回该数组中和为 k 的子数组的个数。子数组是数组中元素的连续非空序列。
阅读更多2024-10-20
「漏洞复现」满客宝智慧食堂系统 selectUserByOrgId 未授权访问漏洞
请勿利用文章内的相关技术从事非法测试，由于传播、利用此文所提供的信息而造成的任何直接或者间接的后果及损失，均由使用者本人负责，作者不为此承担任何责任。工具来自网络，安全性自测，如有侵权请联系删除。本次
阅读更多2024-10-20
React面试题目（从基本到高级）
React前端面试常见题目涵盖了React的基础概念、组件、状态管理、生命周期、性能优化等多个方面。
阅读更多2024-10-20
12.个人博客系统（Java项目基于spring和vue）
1 在校学习的学生，可用于日常学习使用或是毕业设计使用 2 毕业一到两年的开发人员，用于锻炼自己的独立功能模块设计能力，增强代码编写能力。 3 亦可以部署为商化项目使用。 4 需要完整资料及源码
阅读更多2024-10-20
YoloV8改进策略：注意力改进|DeBiFormer，可变形双级路由注意力|引入DeBiLevelRoutingAttention注意力模块（全网首发）
本次改进的核心在于将DeBiLevelRoutingAttention模块嵌入到YoloV8的主干网络中，具体位置是在SPPF（Spatial Pyramid Pooling Fast）模块之后。这一
阅读更多2024-10-20
word取消自动单词首字母大写
情况说明：在word输入单词后首字母会自动变成大写取消单词首字母大写步骤：（1）点击菜单栏文件（2）点击“更多”——>“选项”（3）点击“校对”——>“自动更正选项”（4）取消“句首字母大
阅读更多2024-10-20
web前端网页用户注册页面
【代码】web前端网页用户注册页面。
阅读更多2024-10-20

Java的RocketMQ使用

相关文章