官网
参考:
https://rocketmq.apache.org/zh/docs/quickStart/02quickstart
https://rocketmq.apache.org/zh/docs/deploymentOperations/15deploy
下载地址:
可以选择合适的版本,我选择的是5.0.0版本
1、部署
-
可以使用源码包安装或者二进制包安装,建议下载二进制包,因为是Java开发,需要事先配置jvm。以下只是单台机器的部署做测试,做集群可以参考官网文档。
-
linux基本环境
64位linux系统
64位jdk1.8+=
maven3.2+=
操作步骤
# 自定义存放目录,并上传文件
cd /opt/rocketmq5/
# 解压
unzip rocketmq-all-5.0.0-bin-release.zip
# 添加环境变量
vim /etc/profile
# 增加以下配置
export ROCKETMQ_HOME=/opt/rocketmq5/rocketmq-all-5.0.0-bin-release
export PATH=$ROCKETMQ_HOME/bin:$PATH
source /etc/profile
# 根据需求修改rocketmq启动内存大小,避免内存溢出,如为虚拟机测试可调小一点,生产环境则根据情况调大或调小
# 修改rocketmq安装目录下的文件:bin/runserver.sh、bin/runbroker.sh
runserver.sh修改参考:
JAVA_OPT=”${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m”
JAVA_OPT=”${JAVA_OPT} -server -Xms256m -Xmx256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m”
runbroker.sh修改参考:
JAVA_OPT=”${JAVA_OPT} -server -Xms256m -Xmx256m”
# 启动NameServer
nohup sh /opt/rocketmq5/rocketmq-all-5.0.0-bin-release/bin/mqnamesrv > /opt/rocketmq5/rocketmq-all-5.0.0-bin-release/nohup.out 2>&1 &
# 启动Broker+Proxy
nohup sh /opt/rocketmq5/rocketmq-all-5.0.0-bin-release/bin/mqbroker -n localhost:9876 –enable-proxy > /opt/rocketmq5/rocketmq-all-5.0.0-bin-release/broker.out 2>&1 &
# 关闭rocketmq服务
#关闭proxy
sh /opt/rocketmq5/rocketmq-all-5.0.0-bin-release/bin/mqshutdown proxy
# 关闭NameServer
sh /opt/rocketmq5/rocketmq-all-5.0.0-bin-release/bin/mqshutdown namesrv
#JPS命令查看java进程
jps
# 测试
# 生产消息
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
# 消费消息
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
没有报错就算安装成功。
RocketMQ Dashboard安装参考链接:https://rocketmq.apache.org/zh/docs/deploymentOperations/18Dashboard
注意:如果主机访问不了虚拟机的RocketMQ Dashboard,可以检查虚拟机8080、8081、9876,10911,11011 端口端口是否开放
@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = RocketMqConstants.DISTRIBUTION_ORDER_TOPIC,
consumerGroup = RocketMqConstants.DISTRIBUTION_ORDER_TOPIC)
public class DistributionOrderConsumer implements RocketMQListener<OrderConsumerDTO> {
@Override
public void onMessage(OrderConsumerDTO orderConsumerDTO) {
log.info("开始消费消息,消费信息为:{}", orderConsumerDTO);
HxTenantContextHolder.setCurrentTenantId(orderConsumerDTO.getTenantId());
OrderInfo orderInfo = orderInfoService.getById(orderConsumerDTO.getOrderId());
if (ObjectUtil.isNull(orderInfo)) {
return;
}
long count = distributionOrderService
.count(Wrappers.<DistributionOrder>lambdaQuery().eq(DistributionOrder::getOrderId, orderInfo.getId()));
if (count <= 0) {
List<OrderItem> orderItemList = orderItemService
.list(Wrappers.<OrderItem>lambdaQuery().eq(OrderItem::getOrderId, orderInfo.getId()));
UserInfo userInfo = userInfoService.getById(orderInfo.getUserId());
if (ObjectUtil.isNotNull(userInfo) && StrUtil.isNotBlank(userInfo.getParentId())) {
if (distributionConfigService.isDistribution(orderInfo, userInfo)) {
distributionOrderService.saveDistribution(orderInfo, orderItemList, userInfo, 1);
}
}
}
}
}