04.03 RocketMQ 消息队列

### JooLun支持版本：V1.1.0

部署RocketMQ

• 版本RocketMQ5.3.2
• 方法一：本地部署参考官方：https://rocketmq.apache.org/zh/docs/quickStart/01quickstart
• 方法二：Docker Compose 部署

1. docker-compose.yml文件

version: '3.8'
services:
  namesrv:
    image: apache/rocketmq:5.3.2
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    command: sh mqnamesrv

  broker:
    image: apache/rocketmq:5.3.2
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
      - 10912:10912
    environment:
      - NAMESRV_ADDR=rmqnamesrv:9876
    volumes:
      - ./rocketmq/broker/conf/broker.conf:/etc/rocketmq/broker.conf
    depends_on:
      - namesrv
    command: ["sh","mqbroker","-c","/etc/rocketmq/broker.conf"]

  dashboard:
    image: apacherocketmq/rocketmq-dashboard:latest
    container_name: rmqdashboard
    depends_on:
      - rmqnamesrv
    ports:
      - 8082:8082
    environment:
      - JAVA_OPTS= -Xms256M -Xms512M -Xmn128M -Drocketmq.namesrv.addr=rmqnamesrv:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false
    restart: unless-stopped

2. 在docker-compose.yml同目录下创建配置文件：/rocketmq/broker/conf/broker.conf，配置brokerIP1； docker环境默认使用的docker内部IP，当在docker以外环境使用rocketmq时会超时报错：com.alibaba.rocketmq.remoting.exception.RemotingConnectException: connect to <192.168.0.45:10909> failed；所以要通过过brokerIP1设置宿主机IP，不使用docker内部IP

#启动IP,你机器的实现IP
brokerIP1=192.168.31.100

3. 启动 docker-compose up -d
4. 配置环境的hosts

127.0.0.1 joolun-plus-nacos 
127.0.0.1 joolun-plus-gateway 
127.0.0.1 joolun-plus-redis 
127.0.0.1 joolun-plus-mysql
192.168.31.100 joolun-plus-es
192.168.31.100 joolun-plus-mq-namesrv

JooLun中使用

• 登录nacos后台配置application-dev.yml中的RocketMQ信息

# rocketmq 配置项
rocketmq:
  # Producer 配置项
  producer:
    # 生产者分组
    group: ${spring.application.name}_PRODUCER
  # RocketMQ Namesrv
  name-server: joolun-plus-mq-namesrv:9876

• 服务引用mq模块

	<dependency>
		<groupId>com.joolun</groupId>
		<artifactId>joolun-common-mq</artifactId>
	</dependency>

• 以发送微信模板消息为例：订单提交成功后异步发微信模板消息给用户

1. 在微信服务定义Message消息体TemplateMsgSendMessage

    @Data
    public class TemplateMsgSendMessage implements Serializable {
        private String tenantId;
        /**
         * 商城用户ID
         */
        private String mallUserId;
        /**
         * 用途
         */
        private String useType;
        /**
         * 点击模板卡片后的跳转页面，仅限本小程序内的页面.
         */
        private String page;
        /**
         * 模板内容
         */
        private Map<String, Object> data;
    }

2. 在微信服务定义TemplateMsgMpSendConsumer消费者

    @Component
    @AllArgsConstructor
    @RocketMQMessageListener(consumerGroup = MQConstant.WX_MP_TEMPLATE_MSG_SEND_TOPIC + "_CONSUMER", topic = MQConstant.WX_MP_TEMPLATE_MSG_SEND_TOPIC)
    public class TemplateMsgMpSendConsumer implements RocketMQListener<TemplateMsgSendMessage> {

        private final WxUserService wxUserService;
        private final WxTemplateMsgService wxTemplateMsgService;

        @Override
        public void onMessage(TemplateMsgSendMessage message) {
            TenantContextHolder.setTenantId(message.getTenantId());
            //查询出公众号用户信息，并发送模板消息
            List<WxUser> listWxUser2 = wxUserService.list(Wrappers.<WxUser>lambdaQuery()
                    .eq(WxUser::getMallUserId,message.getMallUserId())
                    .eq(WxUser::getAppType, WxConstant.WX_APP_TYPE_2));
            if(listWxUser2 != null && listWxUser2.size() > 0){
                listWxUser2.forEach(wxUser -> {
                    wxTemplateMsgService.sendWxMpTemplateMsg(message, wxUser);
                });
            }
        }
    }

3. 订单提交成功后调用异步发送模板消息

	@Override
	@Transactional(rollbackFor = Exception.class)
	public OrderInfo orderAddOne(PlaceOrderDTO placeOrderDTO) {
        ......
        //发送微信订阅、模板消息给商城用户，mq异步
        TemplateMsgSendMessage wxTemplateMsgSendDTO = new TemplateMsgSendMessage();
        ......
        wxTemplateMsgSendDTO.setData(data);
        rocketMQTemplate.syncSend(MQConstant.WX_MP_TEMPLATE_MSG_SEND_TOPIC, wxTemplateMsgSendDTO);
        ......
    }