Spring实现RocketMQ简单实例

Spring实现RocketMQ简单实例

RocketMQ 简介

RocketMQ 是一个分布式消息中间件，其具有低延迟、高性能和可靠性、万亿级容量、灵活的可扩展性特性。它主要有四部分组成，分别为 name servers，brokers，producers and consumers。

详细介绍：https://www.cnblogs.com/myitnews/p/12905309.html

RocketMQ安装

前提：

JDK

下载地址：

https://rocketmq.apache.org/release_notes/

启动：

// 解压进入rocketmq的bin目录
// 启动name server
mqnamesrv
// 启动broker
mqbroker -n localhost:9876 autoCreateTopicEnable=true

创建实例主题

mqadmin updateTopic -b localhost:10911 -t my-topic

添加依赖

// https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-spring-boot-starter
implementation group: 'org.apache.rocketmq', name: 'rocketmq-spring-boot-starter', version: '2.2.1'

application.properties

# 定义name-server地址
rocketmq.name-server=127.0.0.1:9876

# 定义发布者组名
rocketmq.producer.group=my-producer-group

Producer组件

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class Producer {

    // 声明并引用RocketMQTemplate
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topicName, String topicContent) {
        rocketMQTemplate.asyncSend(topicName, topicContent, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功，SendStatus = " + sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("异步消息发送异常，exception = " + e.getMessage());
            }
        });
    }
}

Consumer组件

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

@Component
@RocketMQMessageListener(topic = "my-topic", consumerGroup = "my-consumer-group")
public class Consumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("consumer received: " + message);
    }
}

Controller实现

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.Map;

@RestController
public class Controller {

    @Autowired
    private Producer producer;

    @PostMapping("/send/{topicName}")
    public void sendProducerMessage(@PathVariable String topicName, @RequestBody Map<String, String> data) {
        producer.sendMessage(topicName, data.get("message"));
    }
}

运行

Spring Boot启动

HTTP请求：

url: http://127.0.0.1:8080/send/my-topic
method: POST
data: {
	"message": "hello consumer! "
}

更多消息生产与消费方法

参考：

https://juejin.cn/post/6850418115382738958