【没有测开，只有测试干开发】我也用上了MQ......

软件测试君

发布于 2024-06-18 16:25:17

610

发布于 2024-06-18 16:25:17

文章被收录于专栏：测试人生测试人生

写在前面

终于到了自己也能写MQ的时候了，一定程度上来说，还是属于解耦操作范畴吧。或许在某些场景的使用上，属于算小才大用了，但是本着以稳为主的打法，防止高并发，这样也可以让线程更好的发挥作用吧。

实战演示

还是沿用点赞这个功能呀，这次我们不用异步处理了，就直接搞MQ了。

1、搭建MQ

1.1、安装Rocketmq

关于mq的安装坑很多，windows系统或许会比Mac容易很多，我尝试用docker装结果被劝退了，真的太坑了，搞了2小时无果，后来就用官方的压缩吧。官网下载地址，我们选择Binary。

1.2、修改配置

下载完后，打开bin目录下runserver.sh，修改下面这部分配置：

JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

修改为：

AVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

打开bin目录下runbroker.sh，修改下面这部分配置：

JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"

修改为：

JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m"

参数介绍：

-Xmx4g 初始堆大小 4g

-Xms4g 最大堆大小4g

-Xmn512m 年轻代大小(1.4or lator)整个JVM内存大小=年轻代大小 + 年老代大小 + 持久代大小。持久代一般固定大小为64m，所以增大年轻代后，将会减小年老代大小。此值对系统性能影响较大，Sun官方推荐配置为整个堆的3/8

1.3、启动 Rocketmq

启动mqnamesrv：nohup sh ./mqnamesrv &再启动mqbroker ：nohup sh ./mqbroker -n localhost:9876 &查看启动日志：cat nohup.out

1.4、关闭 Rocketmq

停止mqbrokersh ./mqshutdown broker停止mqnamesrvsh ./mqshutdown namesrv

1.5、启动成功

image.png

2、后端改造

这里需要先引入依赖，如下：

<!-- RocketMQ-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.2</version>
</dependency>

配置文件增加如下：

server:
  port: 8888

name: test


# 增加数据库连接
spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/wiki?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC
    username: root
    password: 123456
  data:
    # redis配置
    redis:
      host: localhost
      password: Redis000
      port: 6379


# 配置mybatis所有Mapper.xml所在的路径
mybatis:
  mapper-locations: classpath:/mapper/**/*.xml
# 打印所有的sql日志：sql, 参数, 结果
logging:
  level:
    com:
      jiawa:
        wiki:
          mapper: trace


rocketmq:
  name-server: 127.0.0.1:9876 # mq地址
  producer:
    group: default # 必须指定group
    send-message-timeout: 3000 # 消息发送超时时长，默认3s
    retry-times-when-send-failed: 3 # 同步发送消息失败重试次数，默认2
    retry-times-when-send-async-failed: 3 # 异步发送消息失败重试次数，默认2
    customized-trace-topic: TEST__TOPIC

MQ部分，如下：


import com.jiawa.wiki.websocket.WebSocketServer;
import jakarta.annotation.Resource;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;


@Service
@RocketMQMessageListener(consumerGroup = "default", topic = "VOTE_TOPIC")
public class VoteTopicConsumer implements RocketMQListener<MessageExt> {

    private static final Logger LOG = LoggerFactory.getLogger(VoteTopicConsumer.class);

    @Resource
    public WebSocketServer webSocketServer;

    @Override
    public void onMessage(MessageExt messageExt) {
        byte[] body = messageExt.getBody();
        LOG.info("ROCKETMQ收到消息：{}", new String(body));
        webSocketServer.sendInfo(new String(body));
    }
}

service改造如下：


    /*
     * @decription
     * @author longrong.lang
     * @date 2024/2/6 20:57
     * @param docReq
     * @return com.jiawa.wiki.resp.PageResp<com.jiawa.wiki.resp.DocQueryResp>
     */
    public PageResp<DocQueryResp> list(DocQueryReq req) {
        DocExample docExample = new DocExample();
        docExample.setOrderByClause("sort asc");
        DocExample.Criteria criteria = docExample.createCriteria();
        if (!ObjectUtils.isEmpty(req.getName())){
            //相当于sql的like查询
            criteria.andNameLike("%"+req.getName()+"%");
        }
        PageHelper.startPage(req.getPage(), req.getSize());
        List<Doc> docList = docMapper.selectByExample(docExample);

        PageInfo<Doc> pageInfo = new PageInfo<>(docList);
        LOG.info("总行数：{}", pageInfo.getTotal());
        LOG.info("总页数：{}", pageInfo.getPages());

        // List<DocResp> respList = new ArrayList<>();
        // for (Doc doc : docList) {
        //     // DocResp docResp = new DocResp();
        //     // BeanUtils.copyProperties(doc, docResp);
        //     // 对象复制
        //     DocResp docResp = CopyUtil.copy(doc, DocResp.class);
        //
        //     respList.add(docResp);
        // }

        // 列表复制
        List<DocQueryResp> list = CopyUtil.copyList(docList, DocQueryResp.class);

        PageResp<DocQueryResp> pageResp = new PageResp();
        pageResp.setTotal(pageInfo.getTotal());
        pageResp.setList(list);

        return pageResp;
    }

3、效果

image.png

4、遇到的坑

**报错：**RocketMQ 启动出现 RocketMQTemplate 未注入 Bean？导致这个问题的原因是：Springboot-3.0已经放弃了spring.plants自动装配，它被/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports所取代，添加这个文件是为了兼容。

解决方案：

在resources下创建META-INF，然后在META-INF下创建文件：org.springframework.boot.autoconfigure.AutoConfiguration.imports