开发喵星球

RocketMQ介绍

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题
实现高性能，高可用，可伸缩和最终一致性架构。使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ，这边着重介绍RocketMQ的搭建和使用。RocketMQ是阿里研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。

应用场景

消息队列在互联网技术开发中使用非常广泛，主要用于解决应用耦合，消息分发，流量削锋等问题。以下介绍消息队列在实际应用中常用的使用场景（都以用户下单为例）

安装rocketMq

新建一个生产者模块儿，再建一个消费者模块

第四步在getway模块中配置接口映射规则
第五步添加依赖（生产者和消费者他们的rocketmq依赖一样）

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    # 选择适宜你的版本就好了我的若依项目版本是3.6.3
    <version>2.0.2</version>
</dependency>

第六步添加配置

生产者的

# Tomcat
server:
  port: 10301

# Spring
spring: 
  application:
    # 应用名称
    name: mq-producer
  profiles:
    # 环境配置
    active: dev
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 127.0.0.1:9849
      config:
        # 配置中心地址
        server-addr: 127.0.0.1:9849
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-{spring.profiles.active}.{spring.cloud.nacos.config.file-extension}
rocketmq:
  name-server: 192.168.94.130:9876
  producer:
    # 发送同一类消息的设置为同一个group，保证唯一
    group: springboot_producer_group
    # 发送消息超时时间，默认3000
    sendMessageTimeout: 10000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处，默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度，默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值，默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker，默认false
    retryNextServer: false
#    stream:
#      bindings:
#        ## 新版本固定格式  函数名-{out/in}-{index}
#        demoChannel-out-0:
#          destination: stream-test-topic
#      rocketmq:
#        binder:
#          name-server: 192.168.94.130:9876

消费者的

# Tomcat
server:
  port: 10302

# Spring
spring: 
  application:
    # 应用名称
    name: mq-consumer
  profiles:
    # 环境配置
    active: dev
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 127.0.0.1:9849
      config:
        # 配置中心地址
        server-addr: 127.0.0.1:9849
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-{spring.profiles.active}.{spring.cloud.nacos.config.file-extension}
rocketmq:
  consumer:
    group: springboot_consumer_group
    # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
    pull-batch-size: 10
  name-server: 192.168.94.130:9876

逻辑代码：
生产消息的代码

package com.zhj.mqproducer.controller;

import lombok.Setter;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
@RequestMapping("/mqpro")
public class SendMessageController {


    @Setter(onMethod_ = @Autowired)
    private RocketMQTemplate rocketmqTemplate;

    @GetMapping("/test")
    public  String test() {
        Message<String> msg = MessageBuilder.withPayload("Hello,RocketMQ").build();
        rocketmqTemplate.send("eatfan", msg);
        return "haha";
    }

}

消费消息的代码

package com.zhj.mqconsumer.domain;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "eatfan", consumerGroup = "consumer-zhj2")
public class MyConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Received message: " + message);
    }
}

在浏览器地址中输入如图信息发送消息