> 文档中心 > Springboot使用RocketMQ

Springboot使用RocketMQ


一、新建topic

1、访问console控制台

2、进入topic->新建->输入主题名称

pems参数含义(一般情况设置为: 6)
6:同时支持读写
4:禁写
2:禁读
在这里插入图片描述

二、更新配置,创建对应用户

1、进入文件夹:…\rocketmq-4.9.2\conf\plain_acl.yml

39服务器位置:E:\runtools\rocketmq\rocketmq-4.9.2\conf

2、用户权限配置,按需加入对应配置

支持热更新,有watch方法会监听文件改变(500ms一次,比对文件md5值)
注意:账号密码长度要在6位以上

# 1、fas总队消息队列配置- accessKey: XXXXX  secretKey: XXXXX  whiteRemoteAddress:  # 是否是超级管理员  admin: false  # 默认topic权限。该值默认为DENY(拒绝)。  defaultTopicPerm: DENY  # 默认消费组权限,该值默认为DENY(拒绝),建议值为SUB。  defaultGroupPerm: SUB  # 对fas-corps-operate-queue主题消息有推送和订阅权限  topicPerms:  - fas-corps-operate-queue=PUB|SUB  # 对fas4-corps-consumer消费组有推送和订阅权限  groupPerms:  - fas4-corps-consumer=PUB|SUB

三、springboot:pom引入

 <!--注意: 这里的版本,要和部署在服务器上的版本号一致--> <dependency>     <groupId>org.apache.rocketmq</groupId>     <artifactId>rocketmq-client</artifactId>     <version>4.9.2</version> </dependency> <dependency>     <groupId>org.apache.rocketmq</groupId>     <artifactId>rocketmq-acl</artifactId>     <version>4.9.2</version> </dependency>

四、springboot:队列生产者

1、生产者配置

package com.haiot.utils.rocketmq;import com.alibaba.fastjson.JSONObject;import com.haiot.utils.RedisUtil;import com.haiot.vo.rocketmq.CorpsOperateVO;import lombok.extern.slf4j.Slf4j;import org.apache.rocketmq.acl.common.AclClientRPCHook;import org.apache.rocketmq.acl.common.SessionCredentials;import org.apache.rocketmq.client.exception.MQClientException;import org.apache.rocketmq.client.producer.DefaultMQProducer;import org.apache.rocketmq.common.message.Message;import org.apache.rocketmq.remoting.RPCHook;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;import javax.annotation.Resource;import static com.haiot.commen.constant.CommenConstant.REDIS_CORPS_Detry;import static com.haiot.commen.constant.RocketMQConst.Producer.CorpsProducer;import static com.haiot.commen.constant.RocketMQConst.Topic.CorpsTopic;/** * rocketmq生产者 * @Description  * @Author LiuAnmin * @Version  1.0 * @Date 2022/2/25 12:02 */@Component@Slf4jpublic class Producer {    private DefaultMQProducer producer; static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials("accessKey", "secretKey"));    } public Producer() { //示例生产者 producer = new DefaultMQProducer("test", getAclRPCHook()); //不开启vip通道 开通口端口会减2 producer.setVipChannelEnabled(false); //绑定name server producer.setNamesrvAddr("39.98.138.165:9876"); start();    }    /**     * 对象在使用之前必须要调用一次,只能初始化一次     */    private void start() { try {     this.producer.start(); } catch (MQClientException e) {     e.printStackTrace(); }    }    public DefaultMQProducer getProducer() { return this.producer;    }    /**     * 一般在应用上下文,使用上下文监听器,进行关闭     */    private void shutdown() { this.producer.shutdown();    }}

2、生产者使用

    @Resource    private Producer producer;    @Test    void queueCommit(){ CorpsOperateVO corpsOperateVO = new CorpsOperateVO(); corpsOperateVO.setOperateUrl("aaaa"); corpsOperateVO.setBody("adsfasdfasdf"); //创建生产信息 Message message = new Message("fas-corps-operate-queue", "test", JSONObject.toJSONString(corpsOperateVO).getBytes()); try {     producer.getProducer().send(message); } catch (Exception e) {     e.printStackTrace(); }    }

五、springboot:队列消费者

package com.haiot.utils.rocketmq;import com.alibaba.fastjson.JSONObject;import com.haiot.utils.EmailUtil;import com.haiot.utils.RedisUtil;import com.haiot.utils.RestTemplateUtil;import com.haiot.vo.corps.CorpsResponseVO;import com.haiot.vo.rocketmq.CorpsOperateVO;import lombok.extern.slf4j.Slf4j;import org.apache.rocketmq.acl.common.AclClientRPCHook;import org.apache.rocketmq.acl.common.SessionCredentials;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;import org.apache.rocketmq.client.exception.MQClientException;import org.apache.rocketmq.common.consumer.ConsumeFromWhere;import org.apache.rocketmq.common.message.MessageExt;import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;import org.apache.rocketmq.remoting.RPCHook;import org.springframework.stereotype.Component;import javax.annotation.Resource;import java.nio.charset.StandardCharsets;import static com.haiot.commen.constant.CommenConstant.REDIS_CORPS_Detry;import static com.haiot.commen.constant.RocketMQConst.Consumer.CorpsConsumer;import static com.haiot.commen.constant.RocketMQConst.Topic.CorpsTopic;@Slf4j@Componentpublic class CorpsConsumer {    private DefaultMQPushConsumer consumer;    /**     * 鉴权工作,返回ACL配置     * @Description     * @Author LiuAnmin     * @Version  1.0     * @Date 2022/2/25 12:01     */    static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials("accessKey", "secretKey"));    }    /**     * 通过构造函数 实例化对象     */    public CorpsConsumer() { //消息队列分配算法AllocateMessageQueueAveragely consumer = new DefaultMQPushConsumer("consumer-group", getAclRPCHook(), new AllocateMessageQueueAveragely()); consumer.setNamesrvAddr("39.98.138.165:9876"); //设置消费者的消费模式:广播模式:所有客户端接收的消息都是一样的;默认的模式:负载均衡 consumer.setMessageModel(MessageModel.CLUSTERING); //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); //订阅主题和 标签( * 代表所有标签)下信息 try {     consumer.subscribe(CorpsTopic, "*"); } catch (MQClientException e) {     e.printStackTrace(); } //注册消费的监听 并在此监听中消费信息,并返回消费的状态信息 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {     //遍历消费消息     msgs.forEach(msg -> {     String body = new String(msg.getBody(), StandardCharsets.UTF_8);  log.info("打印队列消息:{}",body);});     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }); try {     consumer.start(); } catch (Exception e) {     e.printStackTrace(); } log.info("消费者启动完毕!");    }}

六、springboot:消费结果输出

Springboot使用RocketMQ

七、更多生产、消费模式查看

查看更多生产、消费模式实际使用

手机爆料