springboot集成商业paas版rocketmq

guaotianxia

949人浏览 · 2023-01-03 10:40:47

guaotianxia · 2023-01-03 10:40:47 发布

一直秉着项目中使用开源软件的原则，期间也遇到过开源软件bug，所以在项目生产环境也尽量不要使用最新版开源软件。有幸参与土豪公司项目，使用了商业版的rocketMq，与开源的使用有点差别，直接上干货记录下

1、pom.xml引用

        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>ons-client</artifactId>
            <version>1.8.8.1.Final</version>
        </dependency>

2、MessageProperties.java

package com.wfl.user.biz.modules.rocketmq.config;

import com.aliyun.openservices.ons.api.PropertyKeyConst;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * 属性配置
 *
 * @author wangfenglei
 */
@Data
@Configuration
public class MessageProperties {
    @Value("${rocketmq.name-server}")
    private String namesrvAddr;
    @Value("${rocketmq.accessKey}")
    private String accessKey;
    @Value("${rocketmq.secretKey}")
    private String secretKey;
    @Value("${rocketmq.instanceId}")
    private String instanceId;
    @Value("${rocketmq.group}")
    private String groupId;

    /**
     * 获取配置
     * @return 配置
     */
    public Properties getMqProperties() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
        properties.put(PropertyKeyConst.NAMESRV_ADDR, namesrvAddr);
        //实例Id
        properties.put(PropertyKeyConst.INSTANCE_ID, instanceId);
        return properties;
    }

}

用于获取rocket的相关配置，可以配置到application.yml文件中或者nacos的配置文件中

3、RocketmqConfig.java

package com.wfl.user.biz.modules.rocketmq.config;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.wfl.user.biz.modules.rocketmq.consumer.listener.BaseAliyunMsgTopicListener ;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * 阿里云pass版rocketMq配置
 *
 * @author wangfenglei
 */
@Configuration
@RefreshScope
public class RocketmqConfig {
    @Autowired
    private MessageProperties messageProperties;

    @Autowired
    private List<BaseAliyunMsgTopicListener> msgTopicListeners;

    /**
     * 注册生产者Bean
     *
     * @return producer
     */
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ProducerBean getProducer() {
        ProducerBean producerBean = new ProducerBean();
        Properties properties = messageProperties.getMqProperties();
        producerBean.setProperties(properties);
        return producerBean;
    }

    /**
     * 获取消息者
     *
     * @return 消费者
     */
    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean getCommonConsumer() {
        ConsumerBean consumerBean = new ConsumerBean();
        Properties properties = messageProperties.getMqProperties();
        //默认为集群模式  测试时可设置为广播
        // properties.put(PropertyKeyConst.MessageModel, MessageModel.BROADCASTING);
        consumerBean.setProperties(properties);

        //组装订阅者消息
        Map<Subscription, MessageListener> map = getMessageListenerMap();
        if (null != map) {
            consumerBean.setSubscriptionTable(map);
        }

        return consumerBean;
    }

    /**
     * 获取消息监听集合
     *
     * @return 监听集合
     */
    private Map<Subscription, MessageListener> getMessageListenerMap() {
        if (null != msgTopicListeners && msgTopicListeners.size() > 0) {
            Map<Subscription, MessageListener> map = new HashMap<>();
            for (BaseAliyunMsgTopicListener msgTopicListener : msgTopicListeners) {
                if (null != msgTopicListener.getTopic()) {
                    Subscription subscription = new Subscription();
                    subscription.setTopic(msgTopicListener.getTopic());
                    subscription.setExpression(msgTopicListener.getTags());
                    map.put(subscription, msgTopicListener);
                }
            }
            if (map.size() > 0) {
                return map;
            }
        }
        return null;
    }

}

用于创建生产者和消费者，在创建消费者时创建监听，每个主题一个消费者

4、BaseAliyunMsgTopicListener.java

package com.wfl.user.biz.modules.rocketmq.consumer.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 消息监听器
 *
 * @author wangfenglei
 */

public abstract class BaseAliyunMsgTopicListener implements MessageListener {
    private static final Logger log = LoggerFactory.getLogger(BaseAliyunMsgTopicListener.class);
    private MsgTagStrategyInterface msgStrategyConsumer;

    /**
     * 消息消息
     *
     * @param message        消息
     * @param consumeContext consumeContext
     * @return 消息结果
     */
    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        try {
            log.info(" ConsumerListener--->Recive mq message success. Topic is:" + message.getTopic() + ", Tag is" + message.getTag() + "msgId is: " + message.getMsgID());

            if (null == msgStrategyConsumer) {
                msgStrategyConsumer = getMsgConsumer();
            }

            Action action = msgStrategyConsumer.consume(message, consumeContext);
            return action;
        } catch (Exception e) {
            log.error(" ConsumerListener--->Consumer mq message fail. Topic is:" + message.getTopic() + ", Tag is" + message.getTag() + "msgId is: " + message.getMsgID(), e);
            return Action.ReconsumeLater;
        }
    }

    /**
     * 获取topic
     *
     * @return topic
     */
    public abstract String getTopic();

    /**
     * 获取tags
     *
     * @return tags
     */
    public abstract String getTags();

    /**
     * 获取消费者
     *
     * @return 消息者
     */
    public abstract MsgTagStrategyInterface getMsgConsumer();
}

消息消费父类，监听收到消息后，此类进行处理

5、FfmAliyunMsgTopicListener.java

package com.wfl.user.biz.modules.rocketmq.consumer.listener.impl;

import com.wfl.user.biz.modules.rocketmq.consumer.aliyun.FfmAliyunMsgConsumer;
import com.wfl.user.biz.modules.rocketmq.consumer.listener.BaseAliyunMsgTopicListener;
import com.wfl.user.biz.modules.rocketmq.consumer.listener.MsgTagStrategyInterface;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * FFM报文监听器
 *
 * @author wangfenglei
 */
@Component
public class FfmAliyunMsgTopicListener extends BaseAliyunMsgTopicListener {
    @Value("${rocketmq.customer.topic.FFM}")
    private String ffmTopic;
    @Value("${rocketmq.customer.topic.FFM.tags}")
    private String ffmTags;
    @Resource
    private FfmAliyunMsgConsumer ffmAliyunMsgConsumer ;

    @Override
    public String getTopic() {
        return ffmTopic;
    }

    @Override
    public String getTags() {
        return ffmTags;
    }

    @Override
    public MsgTagStrategyInterface getMsgConsumer() {
        return ffmAliyunMsgConsumer ;
    }
}

具体的消息监听类，可以根据主题需要，创建多个具体监听类

7、MsgTagStrategyInterface.java

package com.wfl.user.biz.modules.rocketmq.consumer.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;

/**
 * 消息策略类
 * @author wangfenglei
 */
public interface MsgTagStrategyInterface {
    /**
     * 消息消息
     * @param message 消息
     * @param consumeContext 消费环境类
     * @return 消息结果
     */
    Action consume(Message message, ConsumeContext consumeContext);
}

消息消费策略类，在消息监听中调用

7、 FfmAliyunMsgConsumer.java

package com.wfl.user.biz.modules.rocketmq.consumer.aliyun;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.wfl.user.biz.modules.rocketmq.consumer.AbstractConsumer;
import com.wfl.user.biz.modules.rocketmq.consumer.listener.MsgTagStrategyInterface;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * ffm报文主题接收
 *
 * @author wangfenglei
 */
@Slf4j
@Service
public class FfmAliyunMsgConsumer implements MsgTagStrategyInterface {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            String msg = new String(message.getBody(), "UTF-8");
            log.info("ffm receive msg:" + msg);
            /**
              * todo 
              * @消息处理逻辑
             */
            return Action.CommitMessage;
        } catch (Exception e) {
            log.error(e.toString(), e);
            return Action.ReconsumeLater;
        }
    }
}

具体的消息业务处理逻辑，可以根据具体的主题需要，创建多个实现类型

8、RocketmqMsgProducer.java

package com.wfl.user.biz.modules.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * rocketmq消息生产者
 *
 * @author wangfenglei
 */
@Slf4j
@Service
public class RocketmqMsgProducer {
    @Resource
    private MessageConfig messageConfig;

    /**
     * 发送报文
     *
     * @param topic   主题
     * @param message 消息
     * @return 同步发送报文
     */
    public SendResult syncSend(String topic, String message) {
        String[] topicTagArr = topic.split(":");
        SendResult sendResult = new SendResult();
        sendResult.setSendStatus(SendStatus.SEND_OK);

        try {
            sendMessage(topicTagArr[0], topicTagArr.length > 1 ? topicTagArr[1] : null, message);
        } catch (Exception e) {
            log.error(e.toString(), e);
            sendResult.setSendStatus(SendStatus.FLUSH_DISK_TIMEOUT);
        }

        return sendResult; 
    } 

       /**
     * 普通发送消息
     *
     * @param topic 主题
     * @param tag   多个用| * 代表所有
     * @param body
     */
    private SendResult sendMessage(String topic, String tag, String body) throws Exception {
        Producer producer = messageConfig.getProducer();
        Message msg = new Message(topic, tag, body.getBytes());
        msg.setKey(IdWorker.getIdStr());

        SendResult sendResult = producer.send(msg);

        if (null != sendResult) {
            log.info(" Send rocketmq message success. Topic is:" + topic + ",Tag is:" + tag + ",msgId is: " + sendResult.getMessageId());
            return sendResult;
        }

        return null;
    }

}

乐奇 Rokid 开放社区

更多推荐

镜游小导：运行在 Rokid AR 眼镜上的景区实景讲解智能体

项目以 RokidAIUI 开放智能体形态交付：页面用.ink描述，运行在眼镜端智能体运行时，直接调用系统能力。AGENTS.md身份镜游小导v1.0.0，景区 AR 实景讲解智能体；权限（Capabilities）microphonenetworkaudiogyroscope；技能（Skills）。