1：下载rocketMq的安装包
下载后之后，解压到指定的目录,我下载的是4.9.3版本的如图
2：配置环境变量（如图）
3：配置完以后启动
进入bin目录，然后双击mqnamesrv.cmd启动，如图就启动成功
4：启动broker
使用cmd进入bin目录，然后执行
start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true
成功如图，切记，一定一定要出现后面的127.0.0.1：9876,出不来那么后面使用会报错
5：rocketMq-console可视化工具
下载该压缩包rocketmq-externals-master，然后解压到指定目录如图
然后修改配置文件，进入到如下图的目录然后进行修改。
6:启动可视化工具
cmd进入到rocketmq-console的目录后执行编译命令执行 mvn cleanpackage-Dmaven.test.skip=true 初次执行可能稍等1到3分钟，
执行完成以后 rocketmq-console目录下就会有一个target文件如图
cmd进入到这个目录，然后执行命令 java -jar rocketmq-console-ng-1.0.1.jar 就会出现如下图
出现这样子就说明启动成功了，然后localhost:8888再浏览器访问如下图
在这里说下顺序，首先先启动mqnamesrv.cmd，然后再启动mqbroker,然后再启动可视化工具，
下面我说下再springboot中的应用
1：打开idea然后新建springboot工程
新建项目-选择sdk然后下一步创建名称和存放的目录，然后点击完成，如下图
然后再新建模块跟上面创建项目一个流程，创建好就如下图
修改pom.xml文件如下图
然后再修改模块里面的pom.xml如图，里面需要依赖的包代码也贴上了。
org.springframework.boot spring-boot-starter-parent2.3.12.RELEASE 4.0.0 org.example rocketMqpom 1.0-SNAPSHOT springbootMq TestMq org.springframework.boot spring-boot-dependencies2.3.12.RELEASE pom import dev true dev
然后我们再创建包，然后创建类，这里先创建一个生产者的监听类procuderConfig，如下图 代码也贴上了
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ProcuderConfig {
@Value("${rocketmq.groupName}")
private String groupName;
@Value("${rocketmq.namesrvAddr}")
private String namesrvAddr;
@Bean
public DefaultMQProducer createDefaultMQProducer() {
DefaultMQProducer producer = new DefaultMQProducer(this.groupName);
producer.setNamesrvAddr(this.namesrvAddr);
try {
producer.start();
System.out.println("我是一个好人"+String.format("producer is start ! groupName:[%s],namesrvAddr:[%s]", groupName, namesrvAddr));
} catch (MQClientException e) {
System.out.println(String.format("producer is error {}", e.getMessage(), e));
}
return producer;
}
}
然后我们创建一个controller类如下图，代码随后贴上
package mq.cn.springboot;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.nio.charset.StandardCharsets;
@RestController
public class ProcuderController {
@Autowired
private DefaultMQProducer defaultMQProducer;
@Value("${rocketmq.topic}")
private String topic;
@GetMapping("/send-message")
public String sendMessage() {
for (int i = 0; i < 1; i++) {
Message message = new Message(this.topic, "springboot 整合 rocketmq 发送的消息".getBytes());
try {
defaultMQProducer.send(message);
} catch (Exception e) {
e.printStackTrace();
System.out.println(String.format("producer is error {}", e.getMessage(), e.getStackTrace()));
}
}
return "OK,success";
}
}
以上是生产者的创建，下面我们来创建消费者的类
2：创建一个消费者的监听ConsumerConfig类如下图，代码也贴上了
package mq.cn.springboot;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.List;
@Configuration
public class ConsumerConfig {
@Value("${rocketmq.namesrvAddr}")
private String namesrvAddr;
@Value("${rocketmq.groupName}")
private String groupName;
@Value("${rocketmq.topic}")
private String topic;
@Bean
public DefaultMQPushConsumer getRocketMQConsumer() throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
consumer.setNamesrvAddr(namesrvAddr);
consumer.subscribe(this.topic, "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.println("message body:" + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
return consumer;
}
}
然后我们再创建springboot的启动类如图，代码也贴上
package mq.cn;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.Import;
@SpringBootApplication(exclude = SecurityAutoConfiguration.class)
@EnableDiscoveryClient
@EnableFeignClients
public class MqApplication {
public static void main(String[] args) {
SpringApplication.run(MqApplication.class, args);
}
//启动类接着声明一个负载均衡的请求器@LoadBalanced，还有请求发起的工具RestTemplate
}
然后创建一个配置文件，如下图，内容我都贴上
## 服务名称
spring:
application:
name: rocketmq
## 端口号
server:
port: 8081
## rocketmq 的 nameServer
rocketmq:
namesrvAddr: 127.0.0.1:9876
topic: TopicTest
## 使用服务名称作为创建 DefaultMQProducer 中的 group
groupName: ${spring.application.name}
然后就可以启动工程了，启动成功的如下图
然后再浏览器访问
http://localhost:8081/send-message
如图，然后控制台会打印出消费者收到的消息如图
然后我们再mq的可视化工具界面就可以看到收到的消息，如图
就到这里，不懂的留言，一起交流it知识，我也是菜菜鸟
