安装jdk17rockermq5.0需要jdk11我这里使用jdk17配置系统环境变量ROCKETMQ_HOMED:\work\mmq\rocketmq-all-5.2.0-bin-release编写启动脚本 桌面创建txt文件后复制路径写到txt文件后 修改后缀名 .batD:cd D:\work\mmq\rocketmq-all-5.2.0-bin-release\binstart mqnamesrv.cmdstart mqbroker.cmd -n 127.0.0.1:9876 -c ..\conf\broker.conf启动 会弹出2个黑窗口 说明启动成功Windows版RockerMq5.0服务端的包-包含dashboard的图形化包启动 dashboard 图形化界面访问地址 http://localhost:8080/#/创建一个springboot的项目 集成rockermq引入依赖?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version3.3.0/version relativePath/ !-- lookup parent from repository -- /parent groupIdcom.xulk/groupId artifactIdspring-boot/artifactId version0.0.1-SNAPSHOT/version namespringAI/name descriptionDemo project for Spring Boot/description packagingjar/packaging dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version3.4.2/version typepom/type scopeimport/scope /dependency /dependencies /dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency !-- dependency groupIdcom.xuxueli/groupId artifactIdxxl-job-core/artifactId version3.0.0/version /dependency-- !-- RocketMQ -- dependency groupIdorg.apache.rocketmq/groupId artifactIdrocketmq-spring-boot-starter/artifactId version2.3.2/version !-- 使用与您RocketMQ服务端匹配的版本 -- /dependency dependency groupIdorg.apache.rocketmq/groupId artifactIdrocketmq-client-java/artifactId version5.0.7/version /dependency dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId executions execution goals goalrepackage/goal /goals /execution /executions /plugin /plugins /build repositories repository idspring-milestones/id nameSpring Milestones/name urlhttps://repo.spring.io/milestone/url snapshots enabledfalse/enabled /snapshots /repository /repositories /project创建配置文件logging: config: classpath:logback.xml server: port: 8081 # RocketMQ Config rocketmq: name-server: 127.0.0.1:9876 producer: group: my-consumer_group # 生产者组名消息生产者 普通消息和延迟消息import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.text.SimpleDateFormat; import java.util.Date; /** * 消息生产者 */ Service public class MqProducer { Autowired private RocketMQTemplate rocketMQTemplate; /** * 发送消息 * param topic 消息主题 * param message 消息内容 */ public void sendMessage(String topic, String message) { SimpleDateFormat sdf new SimpleDateFormat(yyyy-MM-dd HH:mm:ss); String currentTime sdf.format(new Date()); rocketMQTemplate.convertAndSend(topic, currentTime message); } //定时/延时消息发送 public void sendDeliveryTimestampMessage( ) { SimpleDateFormat sdf new SimpleDateFormat(yyyy-MM-dd HH:mm:ss); String currentTime sdf.format(new Date()); System.out.println( 发送时间 currentTime ); currentTime currentTime AAAAAAAAAAAAAAAAAAAAAAAAAA; Message message new Message(my-topic, , , currentTime.getBytes()); //time这里是消费者的接收时间单位是毫秒。 Long time System.currentTimeMillis() 15 * 60 * 1000; message.setDeliverTimeMs(time); DefaultMQProducer producer rocketMQTemplate.getProducer(); try { producer.send(message); }catch (Exception e){ System.out.println( 发送异常----{} e.getLocalizedMessage() ); } } }消息消费者package com.xxl.job.executor.mq; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; import java.text.SimpleDateFormat; import java.util.Date; /** * 消息消费者 */ Service RocketMQMessageListener(topic my-topic, consumerGroup my-consumer_group) public class MqConsumer implements RocketMQListenerString { /** * 消费对应 topic 主题的消息 * param message 生产者发送过来的消息 */ Override public void onMessage(String message) { SimpleDateFormat sdf new SimpleDateFormat(yyyy-MM-dd HH:mm:ss); String currentTime sdf.format(new Date()); System.out.println( 消费端 接收时间 currentTime ); System.out.println(消费端 消费消息 message); } }创建一个 Controllerimport jakarta.annotation.Resource; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; RestController public class RocketMqController { Resource MqProducer mqProducer; /** * 发送消息 */ GetMapping(send) public void send( ) { // mqProducer.sendMessage(my-topic, 666666666666668888888); mqProducer.sendDeliveryTimestampMessage(); } }浏览器请求http://localhost:8081/send去mq的可视化页面查询 收到消息注意、注意、注意、注意延迟消息需要等到延迟时间后才会投递才能到对应的topic找到我这里延迟22分钟;刚开始查询不到22分钟后投递到指定的topic里面才查到了rockermq会有1秒的延迟这个延迟时间包括网络、投递过程以及生产者和消费者之间的毫秒差
