尧图网站建设 免费方案咨询
ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

首页/资讯中心/资讯详情

Springboot集成rocketmq快速入门demo

发布时间：2026/9/23 15:36:48 尧图网站建设
Springboot集成rocketmq快速入门demo

一、rocketmq介绍

RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。

二、rocketmq环境搭建

采用docker-compose搭建，具体配置如下

 
version: '3'
services:# rocket mq name serverrmqnamesrv:image: apache/rocketmq:4.9.6restart: alwayscontainer_name: rocket-server# environment:#   JAVA_OPT_EXT: "-server -Xms64m -Xmx64m -Xmn64m"# volumes:# 映射本地目录权限一定要设置为 777 权限，否则启动不成功# - ../volumes/data/rocket/server/logs:/home/rocketmq/logsnetworks:- rocketmqports:- 9876:9876command: sh mqnamesrv# rocket mq brokerrmqbroker:image: apache/rocketmq:4.9.6restart: alwayscontainer_name: rocket-brokervolumes:# 映射本地目录权限一定要设置为 777 权限，否则启动不成功# - ../volumes/data/rocket/broker/logs:/home/rocketmq/logs# - ../volumes/data/rocket/broker/store:/home/rocketmq/store- ./config/broker.conf:/opt/rocketmq-4.9.6/conf/broker.confenvironment:- NAMESRV_ADDR=rmqnamesrv:9876# - JAVA_OPTS:=-Duser.home=/opt- JAVA_OPT_EXT=-server -Xms64m -Xmx64m -Xmn64mdepends_on:- rmqnamesrvnetworks:- rocketmqports:- 10909:10909- 10911:10911command: sh mqbroker -c /opt/rocketmq-4.9.6/conf/broker.conf# rocket console 这个可以不需要rmqdashboard:image: apacherocketmq/rocketmq-dashboard:1.0.0restart: alwayscontainer_name: rocket-dashboardenvironment:- JAVA_OPTS=-Drocketmq.config.namesrvAddr=rmqnamesrv:9876 -Dserver.port=8180 -Drocketmq.config.isVIPChannel=false# - JAVA_OPT_EXT=-Xms128m -Xmx128m -Xmn128mdepends_on:- rmqnamesrvnetworks:- rocketmqports:- 8180:8180networks:rocketmq:driver: bridgestack:driver: bridge
 
运行docker-compose
 
 
docker-compose -f docker-compose-rocketmq.yml -p rocketmq up -d
注：修改 xx/rocketmq/rocketmq_broker/conf/broker.conf中配置brokerIP1为宿主机IP
 
访问地址：http://ip地址:8180f7429e9ea14f7441c79b51fbcbb20cc7.png 
 
三、代码工程
 
pom.xml
 
 
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>springboot-demo</artifactId><groupId>com.et</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>rocketmq</artifactId><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.1.1</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.8.0</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-common</artifactId><version>4.8.0</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><scope>test</scope></dependency></dependencies></project>
 
entity
 
 
package com.et59.rocketmq.entity;/*** @author liuhaihua* @version 1.0* @ClassName Person* @Description todo* @date 2023年12月29日 9:52*/public class Person {private String name;private int age;public String getName() {return name;}public void setName(String name) {this.name = name;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}
}
 
listener
 
 
package com.et59.rocketmq.listener;import com.et59.rocketmq.entity.Person;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;/*** @Author: heyuhua* @Date: 2021/1/8 15:55*/
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "PERSON_ADD")
public class PersonMqListener implements RocketMQListener<Person> {@Overridepublic void onMessage(Person person) {System.out.println("接收到消息，开始消费..name:" + person.getName() + ",age:" + person.getAge());}
}
 
util
 
 
package com.et59.rocketmq.util;import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;/*** RocketMq助手** @Author: heyuhua* @Date: 2020/1/8 10:03*/
@Component
public class RocketMqHelper {/*** 日志*/private static final Logger LOG = LoggerFactory.getLogger(RocketMqHelper.class);/*** rocketmq模板注入*/@Autowiredprivate RocketMQTemplate rocketMQTemplate;@PostConstructpublic void init() {LOG.info("---RocketMq助手初始化---");}/*** 发送异步消息** @param topic   消息Topic* @param message 消息实体*/public void asyncSend(Enum topic, Message<?> message) {asyncSend(topic.name(), message, getDefaultSendCallBack());}/*** 发送异步消息** @param topic        消息Topic* @param message      消息实体* @param sendCallback 回调函数*/public void asyncSend(Enum topic, Message<?> message, SendCallback sendCallback) {asyncSend(topic.name(), message, sendCallback);}/*** 发送异步消息** @param topic   消息Topic* @param message 消息实体*/public void asyncSend(String topic, Message<?> message) {rocketMQTemplate.asyncSend(topic, message, getDefaultSendCallBack());}/*** 发送异步消息** @param topic        消息Topic* @param message      消息实体* @param sendCallback 回调函数*/public void asyncSend(String topic, Message<?> message, SendCallback sendCallback) {rocketMQTemplate.asyncSend(topic, message, sendCallback);}/*** 发送异步消息** @param topic        消息Topic* @param message      消息实体* @param sendCallback 回调函数* @param timeout      超时时间*/public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout) {rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout);}/*** 发送异步消息** @param topic        消息Topic* @param message      消息实体* @param sendCallback 回调函数* @param timeout      超时时间* @param delayLevel   延迟消息的级别*/public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel) {rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout, delayLevel);}/*** 发送顺序消息** @param message* @param topic* @param hashKey*/public void syncSendOrderly(Enum topic, Message<?> message, String hashKey) {syncSendOrderly(topic.name(), message, hashKey);}/*** 发送顺序消息** @param message* @param topic* @param hashKey*/public void syncSendOrderly(String topic, Message<?> message, String hashKey) {LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey);rocketMQTemplate.syncSendOrderly(topic, message, hashKey);}/*** 发送顺序消息** @param message* @param topic* @param hashKey* @param timeout*/public void syncSendOrderly(String topic, Message<?> message, String hashKey, long timeout) {LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey + ",timeout:" + timeout);rocketMQTemplate.syncSendOrderly(topic, message, hashKey, timeout);}/*** 默认CallBack函数** @return*/private SendCallback getDefaultSendCallBack() {return new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {LOG.info("---发送MQ成功---");}@Overridepublic void onException(Throwable throwable) {throwable.printStackTrace();LOG.error("---发送MQ失败---"+throwable.getMessage(), throwable.getMessage());}};}@PreDestroypublic void destroy() {LOG.info("---RocketMq助手注销---");}}
 
配置文件
 
 
server:port: 8088
#rocketmq配置
rocketmq:name-server: 10.11.68.77:9876# 生产者配置producer:isOnOff: on# 发送同一类消息的设置为同一个group，保证唯一group: hyh-rocketmq-groupgroupName: hyh-rocketmq-group# 服务地址namesrvAddr: 10.11.68.77:9876# 消息最大长度 默认1024*4(4M)maxMessageSize: 4096# 发送消息超时时间,默认3000sendMsgTimeout: 3000# 发送消息失败重试次数，默认2retryTimesWhenSendFailed: 2
 
启动类
 
 
package com.et59.rocketmq;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class DemoApplication {public static void main(String[] args) {SpringApplication.run(DemoApplication.class, args);}
}
 
code repository
 
  • https://github.com/Harries/springboot-demo
 
四、测试
 
 
package com.et59.rocketmq;import com.et59.rocketmq.entity.Person;
import com.et59.rocketmq.util.RocketMqHelper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)
@SpringBootTest
public class MQTests {@Autowiredprivate RocketMqHelper rocketMqHelper;@Testpublic void testProducter() throws InterruptedException {for(int i=0;i<1000000;i++){Person person = new Person();person.setName("heyuhua");person.setAge(25);rocketMqHelper.asyncSend("PERSON_ADD", MessageBuilder.withPayload(person).build());Thread.sleep(1000);}}}
 
运行结果
 
 
2024-02-02 15:20:27.101 INFO 11188 --- [ublicExecutor_3] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:28.116 INFO 11188 --- [ublicExecutor_4] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:29.130 INFO 11188 --- [ublicExecutor_5] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:30.131 INFO 11188 --- [ublicExecutor_7] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:31.142 INFO 11188 --- [ublicExecutor_8] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:32.156 INFO 11188 --- [ublicExecutor_9] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:33.167 INFO 11188 --- [blicExecutor_10] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
2024-02-02 15:20:34.182 INFO 11188 --- [ublicExecutor_3] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
 
五、引用
 
  •  https://rocketmq.apache.org/
  •  http://www.liuhaihua.cn/archives/710205.html
 

          
          
            
              网站建设              视觉设计              SEO优化            
            返回列表