Слияние кода завершено, страница обновится автоматически
rocketONS-starter 是一个基于阿里云ONS消息服务的轻量级Spring Boot Starter。它为您提供了一个简单高效的封装，使您能够快速创建RocketMQ生产者和消费者，实现应用之间的消息传递。适用于高并发、高吞吐、低延迟的分布式消息场景。
rocketONS-starter 基于Spring的ApplicationContext容器管理，自动扫描consumer监听器，并注册启动消费者，用于接收和处理来自阿里云ONS服务分发的消息。根据配置文件动态创建消费者和生产者，自定义消费者的启停开关，并自动序列化和解析消息实体。同时，它还支持消息过滤、顺序消息和延时消息等高级功能，满足不同场景的需求。
将以下依赖添加到您的项目中：
<dependency>
<groupId>io.gitee.zhucan123</groupId>
<artifactId>rocket-ons-spring-boot-starter</artifactId>
<version>1.0.8</version>
</dependency>
rocket:
address: http://xxxx
secretKey: xxxx
accessKey: xxxx
topic: xxxx
groupSuffix: GID_
enable: true
delay: 1000
| 参数名 | 类型 | 是否必填 | 默认值 | 描述 |
|---|---|---|---|---|
| accessKey | String | 是 | - | 用于身份认证的AccessKeyId，创建于阿里云账号管理控制台。 |
| secretKey | String | 是 | - | 用于身份认证的AccessKeySecret，创建于阿里云账号管理控制台。 |
| address | String | 是 | - | 设置TCP协议接入点。 |
| groupSuffix | String | 否 | GID_ | 控制台创建的Group ID的前缀，通常以"GID_"开头。 |
| topic | String | 是 | - | 当生产者未指定topic时使用的默认绑定topic。 |
| delay | Integer | 否 | 1000 | 消息发送延迟毫秒数。 |
| enable | Boolean | 否 | true | 是否启用starter。 |
@EnableRocketONS
public class App {
public static void main(String[] args) {
SpringApplication.run(App.class, args);
}
}
@ConsumerListener(tags = "msg_tag", consumers = 2)
@OnsConfiguration(topic = "topic-example", group = "GID_${example.group}")
public class ExampleConsumerListener implements RocketListener<MessageData> {
@Override
public Action consume(Message message, MessageData messageBody, ConsumeContext consumeContext) {
// 处理业务逻辑
return Action.CommitMessage;
}
}
@Service
public class ExampleProducerService {
@Autowired
private RocketMQTemplate rocketMQTemplate;
public void sendMessage(MessageData messageData) {
rocketMQTemplate.syncSend("topic-example:msg_tag", messageData);
}
}
通过注入RocketMQTemplate，使用syncSend方法同步发送消息。方法参数中的字符串格式为topic:tag，表示发送到指定topic并设置消息tag。
使用RocketMQTemplate的syncSendOrderly方法发送顺序消息，确保消费者按照发送顺序进行消息处理。
rocketMQTemplate.syncSendOrderly("topic-example:msg_tag", messageData, orderId);
在发送消息时，通过设置messageDelayLevel参数指定延时级别。
rocketMQTemplate.syncSend("topic-example:msg_tag", messageData, messageDelayLevel);
使用@ConsumerListener注解的tags属性来实现消息过滤。设置对应的tag值，消费者将只会消费带有该tag的消息。
@ConsumerListener(tags = "msg_tag", consumers = 2)
public class ExampleConsumerListener implements RocketListener<MessageData> { ... }
请注意，您需要在生产消息时设置相应的tag。
rocketMQTemplate.syncSend("topic-example:msg_tag", messageData);
除了同步发送消息，RocketMQTemplate还提供了异步发送消息的方法。使用asyncSend方法发送消息，提供一个回调函数来处理发送结果。
rocketMQTemplate.asyncSend("topic-example:msg_tag", messageData, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
// 处理发送成功的逻辑
}
@Override
public void onException(Throwable e) {
// 处理发送失败的逻辑
}
});
广播模式允许您将消息发送给所有消费者。要启用广播模式，需要在消费者监听器中设置broadcast属性为true。
@ConsumerListener(tags = "msg_tag", consumers = 2, broadcast = true)
public class ExampleConsumerListener implements RocketListener<MessageData> { ... }
当消费者处理消息失败时，可以通过返回Action.ReconsumeLater来触发消息重试。您可以在配置文件中设置重试策略。
yamlCopy code
rocket:
retry: 3
delay: 1000
| 参数名 | 类型 | 是否必填 | 默认值 | 描述 |
|---|---|---|---|---|
| retry | Integer | 否 | 3 | 消息重试次数。 |
| delay | Integer | 否 | 1000 | 消息发送延迟毫秒数。 |
默认情况下，rocketONS-starter 使用 Java 序列化和反序列化消息。但您可以通过实现MessageSerializer接口来自定义序列化和反序列化策略。 public class CustomMessageSerializer implements MessageSerializer { @Override public byte[] serialize(Object obj) { // 实现自定义的序列化逻辑 } @Override public T deserialize(byte[] data, Class clazz) { // 实现自定义的反序列化逻辑 } }
然后，在配置文件中指定自定义序列化器。
rocket:
serializer: com.example.CustomMessageSerializer
RocketMQ 支持延迟消息发送，可以在发送时指定延迟级别。要发送延迟消息，请使用 sendDelay 方法。
int delayLevel = 3; // 延迟级别，具体延迟时间需要参考 RocketMQ 的延迟级别配置
rocketMQTemplate.sendDelay("topic-example:msg_tag", messageData, delayLevel);
RocketMQ 支持发送事务消息，可以在发送消息时关联一个本地事务。使用 sendMessageInTransaction 方法发送事务消息。
rocketMQTemplate.sendMessageInTransaction("topic-example:msg_tag", messageData, new LocalTransactionExecuter() {
@Override
public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
// 执行本地事务
// 返回事务状态
}
}, null);
为了方便集成测试，您可以使用 RocketMQTestListener 注解启动一个测试消费者。测试消费者会把收到的消息存储在内存中，以便测试时验证。
@RocketMQTestListener(topics = "topic-example", tags = "msg_tag")
public class ExampleConsumerListener implements RocketListener<MessageData> { ... }
在测试用例中，您可以使用 RocketMQTestListener.getMessages() 方法获取收到的消息。
为了提高系统性能，您可以通过以下方式调优 RocketMQ：
具体参数配置请参考 RocketMQ 官方文档。
Вы можете оставить комментарий после Вход в систему
Неприемлемый контент может быть отображен здесь и не будет показан на странице. Вы можете проверить и изменить его с помощью соответствующей функции редактирования.
Если вы подтверждаете, что содержание не содержит непристойной лексики/перенаправления на рекламу/насилия/вульгарной порнографии/нарушений/пиратства/ложного/незначительного или незаконного контента, связанного с национальными законами и предписаниями, вы можете нажать «Отправить» для подачи апелляции, и мы обработаем ее как можно скорее.
Комментарии ( 0 )