RocketMQ 5.X下载
《RocketMQ 5.X下载》
下载后,解压缩
修改初始内存
修改runserver.cmd(Linux则为runserver.sh)
将下面的-Xms2g -Xmx2g -Xmn1g调小
if %JAVA_MAJOR_VERSION% lss 17 (
set "JAVA_OPT=%JAVA_OPT% -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
set "JAVA_OPT=%JAVA_OPT% -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
set "JAVA_OPT=%JAVA_OPT% -verbose:gc -Xloggc:"%USERPROFILE%\rmq_srv_gc.log" -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
set "JAVA_OPT=%JAVA_OPT% -XX:-OmitStackTraceInFastThrow"
set "JAVA_OPT=%JAVA_OPT% -XX:-UseLargePages"
set "JAVA_OPT=%JAVA_OPT% %JAVA_OPT_EXT% -cp "%CLASSPATH%""
) else (
set "JAVA_OPT=%JAVA_OPT% -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
rem set "JAVA_OPT=%JAVA_OPT% -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
rem set "JAVA_OPT=%JAVA_OPT% -verbose:gc -Xloggc:"%USERPROFILE%\rmq_srv_gc.log" -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
set "JAVA_OPT=%JAVA_OPT% -XX:-OmitStackTraceInFastThrow"
set "JAVA_OPT=%JAVA_OPT% -XX:-UseLargePages"
set "JAVA_OPT=%JAVA_OPT% %JAVA_OPT_EXT% -cp "%CLASSPATH%""
)
修改runbroker.cmd(Linux则为runbroker.sh)
将下面的-Xms2g -Xmx2g调小
if %JAVA_MAJOR_VERSION% lss 17 (
set "JAVA_OPT=%JAVA_OPT% -server -Xms2g -Xmx2g"
set "JAVA_OPT=%JAVA_OPT% -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0 -XX:SurvivorRatio=8"
set "JAVA_OPT=%JAVA_OPT% -verbose:gc -Xloggc:%USERPROFILE%\mq_gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintAdaptiveSizePolicy"
set "JAVA_OPT=%JAVA_OPT% -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
set "JAVA_OPT=%JAVA_OPT% -XX:-OmitStackTraceInFastThrow"
set "JAVA_OPT=%JAVA_OPT% -XX:+AlwaysPreTouch"
set "JAVA_OPT=%JAVA_OPT% -XX:MaxDirectMemorySize=15g"
set "JAVA_OPT=%JAVA_OPT% -XX:-UseLargePages -XX:-UseBiasedLocking"
set "JAVA_OPT=%JAVA_OPT% %JAVA_OPT_EXT% -cp %CLASSPATH%"
) else (
set "JAVA_OPT=%JAVA_OPT% -server -Xms2g -Xmx2g"
set "JAVA_OPT=%JAVA_OPT% -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0 -XX:SurvivorRatio=8"
rem set "JAVA_OPT=%JAVA_OPT% -verbose:gc -Xloggc:%USERPROFILE%\mq_gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintAdaptiveSizePolicy"
rem set "JAVA_OPT=%JAVA_OPT% -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
set "JAVA_OPT=%JAVA_OPT% -XX:-OmitStackTraceInFastThrow"
set "JAVA_OPT=%JAVA_OPT% -XX:+AlwaysPreTouch"
set "JAVA_OPT=%JAVA_OPT% -XX:MaxDirectMemorySize=15g"
set "JAVA_OPT=%JAVA_OPT% -XX:-UseLargePages -XX:-UseBiasedLocking"
set "JAVA_OPT=%JAVA_OPT% %JAVA_OPT_EXT% -cp "%CLASSPATH%""
)
修改broker.conf
追加以下内容
# 设置NameServer地址和端口
namesrvAddr = 127.0.0.1:9876
# 是否启动自动创建主题,如果自动创建主题被启用,broker会自动创建topic,clusterTopic可以被非管理员用户创建
autoCreateTopicEnable = true
# 是否启动自动创建订阅组,如果启动自动创建订阅组,broker会自动创建订阅组
autoCreateSubscriptionGroup = true
创建Windows启动脚本
RocketMQ_start.bat
@echo off
echo 如果JAVA_HOME和RocketMQ不一致,则可以像下面这样在脚本里面设置 RocketMQ 可以运行的JDK版本
#set JAVA_HOME=C:/Users/X/.jdks/corretto-1.8.0_402
echo 设置 RocketMQ 的bin目录
set rocketmqLocation=D:\softwareWork\rocketmq-all-5.2.0-bin-release\bin\
echo 设置 broker 引用的配置文件
start /min cmd /k %rocketmqLocation%mqbroker.cmd -c ../conf/broker.conf
start /min cmd /k %rocketmqLocation%mqnamesrv.cmd
与SpringBoot集成
pom.xml引入rocketmq
<!-- RocketMQ -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.1</version>
</dependency>
application.yml配置rocketmq 服务地址
# RocketMQ Config
rocketmq:
name-server: 127.0.0.1:9876
创建MqProducer.java消息生产者
一般情况下生产者和消费者是2个不同的服务,这里只是演示demo,因此不做区分
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
/**
* 消息生产者(一般情况,生产者和消费者不在同一个项目)
* 创建日期:2024-05-11
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class MqProducer {
private final RocketMQTemplate rocketMQTemplate;
/**
* 发送消息
* @param topic 消息主题
* @param message 消息内容
*/
public void sendMessage(String topic, String message) {
rocketMQTemplate.convertAndSend(topic, message);
}
}
创建MqConsumer.java消息消费者
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
/**
* 消息消费者(一般情况,生产者和消费者不在同一个项目)
* 创建日期:2024-05-11
*/
@Slf4j
@Service
@RocketMQMessageListener(topic = "my-topic", consumerGroup = "my-consumer_group")
public class MqConsumer implements RocketMQListener<String> {
/**
* 消费对应 topic 主题的消息
* @param message 生产者发送过来的消息
*/
@Override
public void onMessage(String message) {
// 处理接收到的消息
log.info("接收到topic = my-topic的消息,内容为: {}", message);
}
}
创建RocketMqController.java发生消息控制类
import lombok.RequiredArgsConstructor;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
* 创建日期:2024-05-11
*/
@RestController
@RequiredArgsConstructor
public class RocketMqController {
private final MqProducer mqProducer;
/**
* 发送消息
* @param topic 消息主题
* @param message 消息内容
*/
@GetMapping("send")
public void send(@RequestParam(name = "topic", required = false) String topic, @RequestParam(name = "message") String message) {
if (!StringUtils.hasText(topic)){
topic = "my-topic";
}
mqProducer.sendMessage(topic, message);
}
}
启动项目,访问:localhost:8080/send?message=‘haha’
生成者会发生一条topic = "my-topic"主题的消息到RocketMQ,
消费者会消费topic = "my-topic"主题的一条消息,上面的代码就只是简单的把消息输出到控制台,因此控制台会输出:haha