当前位置：首页 > article >正文

【tower-boot 系列】开源RocketMQ和阿里云rockerMq 4.x和5.x集成 (一)

article 2024/11/19 20:30:56

RocketMQ 简单介绍

RockerMq

阿里云rockerMq 4.x和5.x集成

一、云平台创建实例

参考文档：
阿里云api

阿里云创建实例
阿里云Rocker实例

二、skd集成思路

公司用的RocketMQ一般是自建开源apache的RocketMQ和上阿里云的RocketMQ，目前阿里云支持4.x和5.x版本
项目集成思路：
1、集成阿里RocketMQ 两个版本 4.x和5.x 支持版本配置和开源apache的RocketMQ集成
2、RocketProperties 单配置适用多版本集成
3、RocketConsumer 消费者注解，支持多版本集成(消费组监听器继承实现可以不一样，注解一致）
4、RocketMessageProducer 生产者接口，支持多版本集成，根据配置版本自动适配

注：这个是一个持续集成优化过程，需要不停磨练

三、根据RocketConsumer注解，动态监听器实现

1、核心逻辑代码:
扫描包注解，根据配置版本号走不同的，消费组创建

        String packageName = properties.getPackageName(); // 扫描包路径
        ClassPathScanningCandidateComponentProvider scanner = new ClassPathScanningCandidateComponentProvider(false);
        scanner.addIncludeFilter(new AnnotationTypeFilter(RocketConsumer.class));
        Set<BeanDefinition> candidates = scanner.findCandidateComponents(packageName);
        for (BeanDefinition beanDefinition : candidates) {
            try {
                Integer versions = properties.getVersions();
                if (Objects.equals(RockerMqVersions.ALI_4.getCode(), versions)) {
                    onsRocketMQConsumerClient.start(beanDefinition);
                }

                if (Objects.equals(RockerMqVersions.ALI_5.getCode(), versions)) {
                    rocketMQConsumerClient.start(beanDefinition);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

2、消费组push 用法示例
版本4.x 写法

@Slf4j
@Component
@RocketConsumer(topic = "PRODUCER_TOPIC", consumerGroup = "PRODUCER_GROUP")
public class Push4MQConsumer implements MessageListener {

    @Autowired
    private UserService userService;

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        String body = new String(message.getBody());
        log.info("TestMQConsumer:" + body + "user:" + userService.getUserId());
        return Action.CommitMessage;
    }
}

版本5.x 写法

@Slf4j
@Component
@RocketConsumer(topic = "PRODUCER_TOPIC", consumerGroup = "PRODUCER_GROUP")
public class TestMQConsumer implements MessageListener {

    @Autowired
    private UserService userService;

    @Override
    public ConsumeResult consume(MessageView messageView) {
        String body = StandardCharsets.UTF_8.decode(messageView.getBody()).toString();
        log.info("TestMQConsumer:" + body + "user:" + userService.getUserId());
        return ConsumeResult.SUCCESS;
    }
}

五、RocketMessageProducer 生产发送

目前只写5.x 写法,后面持续优化
普通消息

RocketMsg rocketMsg = new RocketMsg();
rocketMsg.setBody("hxl测试发送");
rocketMessageProducer.sendMessage("PRODUCER_TOPIC", null, rocketMsg);
return ApiResult.success();

延迟消息

RocketMsg rocketMsg = new RocketMsg();
rocketMsg.setBody("延迟消息发送发送");
rocketMessageProducer.sendMessage("DELAY_TOPIC", null, 5 * 60L, rocketMsg);

# github 地址：
[https://github.com/sanxiaoshitou/tower-boot](https://github.com/sanxiaoshitou/tower-boot)