Spring Boot 集成 RocketMQ

创建项目，引入依赖

先创建一个简单的Spring Boot项目

引入依赖

xml

<properties>
    <java.version>1.8</java.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <spring-boot.version>2.6.13</spring-boot.version>
    <rocketmq-v5-client.version>2.3.1</rocketmq-v5-client.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-v5-client-spring-boot-starter</artifactId>
        <version>${rocketmq-v5-client.version}</version>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>${spring-boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

这里使用的是RocketMQ 5.0的Spring Boot Starter，需要引入rocketmq-v5-client-spring-boot-starter依赖。

配置

配置文件application.properties

properties

# 应用服务 WEB 访问端口
server.port=8080

rocketmq.producer.endpoints=192.168.118.128:8081
rocketmq.producer.topic=MyTopic

发送消息

方式一

编写一个Controller，用于发送消息

java

package com.example.demo.controller;

import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.core.RocketMQClientTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@RestController
@RequestMapping("/producer")
public class ProducerController {

    @Resource
    private RocketMQClientTemplate rocketMQClientTemplate;

    @Value("${rocketmq.producer.topic}")
    private String topic;

    @PostConstruct
    public void init() {
        System.out.println("主题: " + topic);
    }

    @PostMapping("/t0")
    public String send() {
        SendReceipt sendReceipt = rocketMQClientTemplate.syncSendNormalMessage(topic, "Hello V5");
        System.out.println(sendReceipt);
        return sendReceipt.getMessageId().toString();
    }
}

启动后，调用接口 http://localhost:8080/producer/t0，查看控制台输出

方式二

使用ProducerBuilder创建一个Producer，然后发送消息

java

package com.example.demo.config;

import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.ProducerBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

@Configuration
public class MessageConfig {
    @Resource
    private ProducerBuilder producerBuilder;

    @Bean
    public Producer myProducer() {
        Producer build = null;
        try {
            build = producerBuilder.build();
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
        return build;
    }
}

注入使用

java

package com.example.demo.controller;

import com.example.demo.util.MessageUtil;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.core.RocketMQClientTemplate;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;

@RestController
@RequestMapping("/producer")
public class ProducerController {


    @Resource
    private Producer myProducer;

    @Value("${rocketmq.producer.topic}")
    private String topic;


    @PostMapping("/t2")
    public String send2() {
        MessageBuilder messageBuilder = new MessageBuilderImpl();
        Message message = messageBuilder.setTopic(topic).setBody("Hello V5".getBytes(StandardCharsets.UTF_8))
                .build();
        SendReceipt sendReceipt = null;
        try {
            sendReceipt = myProducer.send(message);
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
        return sendReceipt.getMessageId().toString();
    }
}

消费消息

方式一

使用simple-consumer

配置

properties

# 应用服务 WEB 访问端口
server.port=8080

# 发送
rocketmq.producer.endpoints=192.168.118.128:8081
rocketmq.producer.topic=spring-boot-test-topic-0

# 消费方式1
rocketmq.simple-consumer.endpoints=192.168.118.128:8081
rocketmq.simple-consumer.topic=spring-boot-test-topic-0
rocketmq.simple-consumer.tag=*
rocketmq.simple-consumer.consumer-group=spring-boot-test-consumer-group-0

接收消息

java

package com.example.demo.controller;

import com.example.demo.util.MessageUtil;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.core.RocketMQClientTemplate;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;

@RestController
@RequestMapping("/producer")
public class ProducerController {

    @Resource
    private RocketMQClientTemplate rocketMQClientTemplate;

    @Value("${rocketmq.producer.topic}")
    private String topic;

    @GetMapping("/t1")
    public String getMessage() {
        List<MessageView> receive = null;
        try {
            receive = rocketMQClientTemplate.receive(1, Duration.ofSeconds(10));
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
        for (MessageView messageView : receive) {
            ByteBuffer body = messageView.getBody();
            String message = MessageUtil.getMessage(body);
            System.out.println(messageView.getMessageId()+ ": " + message);
        }
        return "Ok";
    }
}

方式二

使用注解RocketMQMessageListener

配置

properties

# 消费方式2
rocketmq.push-consumer.endpoints=192.168.118.128:8081
rocketmq.push-consumer.topic=spring-boot-test-topic-0
rocketmq.push-consumer.consumer-group=spring-boot-test-consumer-group-0
rocketmq.push-consumer.tag=*

具体的配置项见注解RocketMQMessageListener

java

package com.example.demo.listener;

import com.example.demo.util.MessageUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.annotation.RocketMQMessageListener;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.core.RocketMQListener;
import org.springframework.stereotype.Component;


@Slf4j
@Component
@RocketMQMessageListener(
        consumerGroup = "${rocketmq.push-consumer.consumer-group}"
)
public class MessageListener implements RocketMQListener {

    @Override
    public ConsumeResult consume(MessageView messageView) {
        MessageId messageId = messageView.getMessageId();
        String string = MessageUtil.getMessage(messageView.getBody());
        log.info("收到消息id: messageId={}, 消息内容: string={}", messageId.toString(), string);
        return ConsumeResult.SUCCESS;
    }

}

java

package com.example.demo.util;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class MessageUtil {
    public static String getMessage(ByteBuffer byteBuffer) {
        int length = byteBuffer.limit() - byteBuffer.position();
        byte[] bytes = new byte[length];
        byteBuffer.get(bytes);
        return new String(bytes, StandardCharsets.UTF_8);
    }
}

这种方式，消息一到就处理，不用主动调用receive方法

Spring Boot 集成 RocketMQ ​

创建项目，引入依赖 ​

配置 ​

发送消息 ​

方式一 ​

方式二 ​

消费消息 ​

方式一 ​

方式二 ​

Spring Boot 集成 RocketMQ

创建项目，引入依赖

配置

发送消息

方式一

方式二

消费消息

方式一

方式二