liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

底层流量数据改为bit，测试金山云服务器API，准备1.1分支

Browse Source
This commit is contained in:
gaoyutao
2025-10-09 17:38:24 +08:00
parent ab578ee17f
commit 7b5c782079
16 changed files with 211 additions and 655 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/util/CalculateUtil.java
+71 -71
View File
@@ -29,7 +29,7 @@ public class CalculateUtil {
// 处理纯数字（单位为空）
if (unit.isEmpty()) {
unit = "B/S"; // 默认单位: Bytes/s
unit = "B/S"; // 默认单位: Bit/s
}
switch (mode) {
case "1000":
@@ -41,88 +41,88 @@ public class CalculateUtil {
}
}
// 1000进制换算（最小单位为bit）
// private static BigDecimal convertWithDecimalCoefficients(BigDecimal value, String unit) {
// switch (unit) {
// case "B/S": // bit per second
// return value.divide(new BigDecimal("1000000000"), 2, RoundingMode.HALF_UP); // 10^9
// case "KB/S": // Kilobit per second
// return value.divide(new BigDecimal("1000000"), 2, RoundingMode.HALF_UP); // 10^6
// case "MB/S": // Megabit per second
// return value.divide(new BigDecimal("1000"), 2, RoundingMode.HALF_UP); // 10^3
// case "GB/S": // Gigabit per second
// return value.setScale(2, RoundingMode.HALF_UP); // 10^0
// case "TB/S": // Terabit per second
// return value.multiply(new BigDecimal("1000")) // 10^3
// .setScale(2, RoundingMode.HALF_UP);
// default:
// throw new IllegalArgumentException("Unsupported DECIMAL unit: " + unit);
// }
// }
// // 1024进制换算（最小单位为bit）
// private static BigDecimal convertWithBinaryCoefficients(BigDecimal value, String unit) {
// BigDecimal base1024 = new BigDecimal(1024);
// BigDecimal base1048576 = base1024.multiply(base1024); // 1024^2 = 1048576
// BigDecimal base1073741824 = base1048576.multiply(base1024); // 1024^3 = 1073741824
//
// switch (unit) {
// case "B/S": // bit per second
// return value.divide(base1073741824, 2, RoundingMode.HALF_UP); // 1024^3
// case "KB/S": // Kilobit per second
// return value.divide(base1048576, 2, RoundingMode.HALF_UP); // 1024^2
// case "MB/S": // Megabit per second
// return value.divide(base1024, 2, RoundingMode.HALF_UP); // 1024^1
// case "GB/S": // Gigabit per second
// return value.setScale(2, RoundingMode.HALF_UP); // 1024^0
// case "TB/S": // Terabit per second
// return value.multiply(base1024) // 1024^1
// .setScale(2, RoundingMode.HALF_UP);
// default:
// throw new IllegalArgumentException("Unsupported BINARY unit: " + unit);
// }
// }
// 字节1000进制换算
private static BigDecimal convertWithDecimalCoefficients(BigDecimal value, String unit) {
switch (unit.toUpperCase()) {
case "B/S":
return value.multiply(new BigDecimal("0.000008")) // 8/1000000
.setScale(2, RoundingMode.HALF_UP);
case "KB/S":
return value.multiply(new BigDecimal("0.008")) // 8/1000
.setScale(2, RoundingMode.HALF_UP);
case "MB/S":
return value.setScale(2, RoundingMode.HALF_UP);
case "GB/S":
return value.multiply(new BigDecimal("8000")) // 1000 * 8
.setScale(2, RoundingMode.HALF_UP);
case "TB/S":
return value.multiply(new BigDecimal("8000000")) // 1000 * 1000 * 8
switch (unit) {
case "B/S": // bit per second
return value.divide(new BigDecimal("1000000000"), 2, RoundingMode.HALF_UP); // 10^9
case "KB/S": // Kilobit per second
return value.divide(new BigDecimal("1000000"), 2, RoundingMode.HALF_UP); // 10^6
case "MB/S": // Megabit per second
return value.divide(new BigDecimal("1000"), 2, RoundingMode.HALF_UP); // 10^3
case "GB/S": // Gigabit per second
return value.setScale(2, RoundingMode.HALF_UP); // 10^0
case "TB/S": // Terabit per second
return value.multiply(new BigDecimal("1000")) // 10^3
.setScale(2, RoundingMode.HALF_UP);
default:
throw new IllegalArgumentException("Unsupported DECIMAL unit: " + unit);
}
}
// 字节1024进制换算
// 1024进制换算（最小单位为bit）
private static BigDecimal convertWithBinaryCoefficients(BigDecimal value, String unit) {
switch (unit.toUpperCase()) {
case "B/S":
return value.multiply(new BigDecimal("0.00000762939")) // 8/1048576
.setScale(2, RoundingMode.HALF_UP);
case "KB/S":
return value.multiply(new BigDecimal("0.0078125")) // 8/1024
.setScale(2, RoundingMode.HALF_UP);
case "MB/S":
return value.setScale(2, RoundingMode.HALF_UP);
case "GB/S":
return value.multiply(new BigDecimal("8192")) // 1024 * 8
.setScale(2, RoundingMode.HALF_UP);
case "TB/S":
return value.multiply(new BigDecimal("8388608")) // 1024 * 1024 * 8
BigDecimal base1024 = new BigDecimal(1024);
BigDecimal base1048576 = base1024.multiply(base1024); // 1024^2 = 1048576
BigDecimal base1073741824 = base1048576.multiply(base1024); // 1024^3 = 1073741824
switch (unit) {
case "B/S": // bit per second
return value.divide(base1073741824, 2, RoundingMode.HALF_UP); // 1024^3
case "KB/S": // Kilobit per second
return value.divide(base1048576, 2, RoundingMode.HALF_UP); // 1024^2
case "MB/S": // Megabit per second
return value.divide(base1024, 2, RoundingMode.HALF_UP); // 1024^1
case "GB/S": // Gigabit per second
return value.setScale(2, RoundingMode.HALF_UP); // 1024^0
case "TB/S": // Terabit per second
return value.multiply(base1024) // 1024^1
.setScale(2, RoundingMode.HALF_UP);
default:
throw new IllegalArgumentException("Unsupported BINARY unit: " + unit);
}
}
// 字节1000进制换算
// private static BigDecimal convertWithDecimalCoefficients(BigDecimal value, String unit) {
// switch (unit.toUpperCase()) {
// case "B/S":
// return value.multiply(new BigDecimal("0.000008")) // 8/1000000
// .setScale(2, RoundingMode.HALF_UP);
// case "KB/S":
// return value.multiply(new BigDecimal("0.008")) // 8/1000
// .setScale(2, RoundingMode.HALF_UP);
// case "MB/S":
// return value.setScale(2, RoundingMode.HALF_UP);
// case "GB/S":
// return value.multiply(new BigDecimal("8000")) // 1000 * 8
// .setScale(2, RoundingMode.HALF_UP);
// case "TB/S":
// return value.multiply(new BigDecimal("8000000")) // 1000 * 1000 * 8
// .setScale(2, RoundingMode.HALF_UP);
// default:
// throw new IllegalArgumentException("Unsupported DECIMAL unit: " + unit);
// }
// }
//
// // 字节1024进制换算
// private static BigDecimal convertWithBinaryCoefficients(BigDecimal value, String unit) {
// switch (unit.toUpperCase()) {
// case "B/S":
// return value.multiply(new BigDecimal("0.00000762939")) // 8/1048576
// .setScale(2, RoundingMode.HALF_UP);
// case "KB/S":
// return value.multiply(new BigDecimal("0.0078125")) // 8/1024
// .setScale(2, RoundingMode.HALF_UP);
// case "MB/S":
// return value.setScale(2, RoundingMode.HALF_UP);
// case "GB/S":
// return value.multiply(new BigDecimal("8192")) // 1024 * 8
// .setScale(2, RoundingMode.HALF_UP);
// case "TB/S":
// return value.multiply(new BigDecimal("8388608")) // 1024 * 1024 * 8
// .setScale(2, RoundingMode.HALF_UP);
// default:
// throw new IllegalArgumentException("Unsupported BINARY unit: " + unit);
// }
// }
/**
* 计算百分位值
ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/util/HmacUtils.java
+91
View File
@@ -0,0 +1,91 @@
package com.ruoyi.system.util;
import com.ruoyi.common.core.utils.DateUtils;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
import javax.crypto.Mac;
import javax.crypto.SecretKey;
import javax.crypto.spec.SecretKeySpec;
import java.net.URI;
public class HmacUtils {
private static final String MAC_NAME = "HmacSHA1";
private static final String ENCODING = "UTF-8";
/**
*@param encryptKey 秘钥由中控系统分配
*@param plainText 加密内容：token + 时间戳 + URI
*/
public static String getHmac(String plainText,String encryptKey) {
try{
byte[] dataKey=encryptKey.getBytes(ENCODING);
byte[] dataValue=plainText.getBytes(ENCODING);
//根据给定的字节数组构造一个密钥,第二参数指定一个密钥算法的名称
SecretKey secretKey = new SecretKeySpec(dataKey, MAC_NAME);
// 2、创建MAC对象
Mac mac = Mac.getInstance(MAC_NAME);
// 3、设置密钥
mac.init(secretKey);
// 4、数据加密
byte[] bytes = mac.doFinal(dataValue);
// 5、生成数据
String rs = encodeHex(bytes,false);
return rs;
}catch (Exception e){
throw new IllegalArgumentException(e.getMessage());
}
}
// 数据转16进制编码
public static String encodeHex(final byte[] data, final boolean toLowerCase) {
final char[] DIGITS_LOWER = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
final char[] DIGITS_UPPER = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F' };
final char[] toDigits = toLowerCase ? DIGITS_LOWER : DIGITS_UPPER;
final int l = data.length;
final char[] out = new char[l << 1];
// two characters form the hex value.
for (int i = 0, j = 0; i < l; i++) {
out[j++] = toDigits[(0xF0 & data[i]) >>> 4];
out[j++] = toDigits[0x0F & data[i]];
}
return new String(out);
}
public static void main(String[] args) {
long timestamp = DateUtils.getNowDate().getTime();
String plainText = "efea5f0218c84a24b9fdab3264de3da5" + timestamp + "/supplier/outer/dev/getFlow";
String secretKey = getHmac(plainText, "91a73fd806ab2c005c13b4dc19130a884e909dea3f72d46e30266fe1a1f588d8");
System.out.println(secretKey);
RestTemplate restTemplate = new RestTemplate();
// 设置请求头
HttpHeaders headers = new HttpHeaders();
headers.set("Content-Type", "application/json");
headers.set("TOKEN", "efea5f0218c84a24b9fdab3264de3da5");
headers.set("secret-key", secretKey);
headers.set("timestamps", String.valueOf(timestamp));
// 使用 UriComponentsBuilder 构建URL（推荐）
URI uri = UriComponentsBuilder
.fromHttpUrl("https://ecscm-openapi.ksyun.com/supplier/outer/dev/getFlow")
.queryParam("srmChannel", "1000121954")
.queryParam("startTime", "2025-10-09 00:00:00")
.queryParam("endTime", "2025-10-09 01:00:00")
.build()
.toUri();
HttpEntity<String> entity = new HttpEntity<>(headers);
// GET 请求
ResponseEntity<String> response = restTemplate.exchange(
uri,
HttpMethod.GET,
entity,
String.class
);
System.out.println(response.getBody());
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/DeviceMessageHandler.java
+13 -2
View File
@@ -10,6 +10,7 @@ import com.ruoyi.rocketmq.domain.*;
import com.ruoyi.rocketmq.domain.vo.CollectDataVo;
import com.ruoyi.rocketmq.domain.vo.RspVo;
import com.ruoyi.rocketmq.service.*;
import com.ruoyi.rocketmq.utils.DataProcessUtil;
import com.ruoyi.rocketmq.utils.JsonDataParser;
import com.ruoyi.rocketmq.utils.SwitchJsonDataParser;
import com.ruoyi.system.api.RemoteRevenueConfigService;
@@ -93,6 +94,8 @@ public class DeviceMessageHandler {
private IRmResourceRemoteService rmResourceRemoteService;
@Autowired
private IRmAgentManagementService rmAgentManagementService;
@Autowired
private DataProcessUtil dataProcessUtil;
// 在类中添加
private static final ObjectMapper objectMapper = new ObjectMapper();
@@ -186,6 +189,10 @@ public class DeviceMessageHandler {
interfaces.forEach(iface -> {
iface.setClientId(message.getClientId());
iface.setCreateTime(createTime);
// 发送流量
iface.setOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed()));
// 接收流量
iface.setInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed()));
});
// 批量入库集合
data.setList(interfaces);
@@ -498,13 +505,17 @@ public class DeviceMessageHandler {
// 计算inSpeed
if (switchInfo.getInBytes() != null && tempInfo.getInBytes() != null) {
BigDecimal inDiff = switchInfo.getInBytes().subtract(tempInfo.getInBytes());
switchInfo.setInSpeed(inDiff.divide(divisor, 2, RoundingMode.HALF_UP));
// 字节转为bit
BigDecimal inDiffBit = inDiff.multiply(new BigDecimal(8)).setScale(0, RoundingMode.HALF_UP);
switchInfo.setInSpeed(inDiffBit.divide(divisor, 2, RoundingMode.HALF_UP));
}
// 计算outSpeed
if (switchInfo.getOutBytes() != null && tempInfo.getOutBytes() != null) {
BigDecimal outDiff = switchInfo.getOutBytes().subtract(tempInfo.getOutBytes());
switchInfo.setOutSpeed(outDiff.divide(divisor, 2, RoundingMode.HALF_UP));
// 字节转为bit
BigDecimal outDiffBit = outDiff.multiply(new BigDecimal(8)).setScale(0, RoundingMode.HALF_UP);
switchInfo.setOutSpeed(outDiffBit.divide(divisor, 2, RoundingMode.HALF_UP));
}
}
});
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/DataProcessUtil.java
+36
View File
@@ -140,4 +140,40 @@ public class DataProcessUtil {
return devicesResponse != null ? devicesResponse.getData() : Collections.emptyList();
}
/**
* 将采集到的字节流量转化为比特流量
* 只是简单地将字节值乘以8得到比特值
* @param speed 字节数值字符串
* @return 比特数值字符串
*/
public String bytesToBits(String speed) {
if (speed == null || speed.trim().isEmpty()) {
return "0";
}
String cleanedSpeed = speed.trim();
try {
// 解析字节值并乘以8转换为比特值
long bytesValue = Long.parseLong(cleanedSpeed);
long bitsValue = bytesValue * 8;
return String.valueOf(bitsValue);
} catch (NumberFormatException e) {
// 如果解析失败，尝试使用double类型
try {
double bytesValue = Double.parseDouble(cleanedSpeed);
double bitsValue = bytesValue * 8;
// 如果是整数则返回整数形式，否则返回小数形式
if (bitsValue == (long) bitsValue) {
return String.valueOf((long) bitsValue);
} else {
return String.valueOf(bitsValue);
}
} catch (NumberFormatException ex) {
throw new IllegalArgumentException("Invalid speed format: " + speed);
}
}
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/RocketMQApplication.java
-25
View File
@@ -1,25 +0,0 @@
package com.ruoyi.testrocketmq;
import com.ruoyi.common.security.annotation.EnableCustomConfig;
import com.ruoyi.common.security.annotation.EnableRyFeignClients;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
/**
* 平台管理模块
*
* @author ruoyi
*/
@EnableCustomConfig
@EnableRyFeignClients
@SpringBootApplication
@EnableAsync
public class RocketMQApplication
{
public static void main(String[] args)
{
SpringApplication.run(RocketMQApplication.class, args);
System.out.println("(♥◠‿◠)ﾉﾞ RocketMQ模块启动成功 ლ(´ڡ`ლ)ﾞ");
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/config/ConsumerConfig.java
-64
View File
@@ -1,64 +0,0 @@
package com.ruoyi.testrocketmq.config;
import com.ruoyi.testrocketmq.consumer.RocketMsgListener;
import com.ruoyi.testrocketmq.enums.MessageCodeEnum;
import com.ruoyi.testrocketmq.model.ConsumerMode;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 消费者配置
*/
@RefreshScope
@Configuration
@Slf4j
public class ConsumerConfig {
@Autowired
private ConsumerMode consumerMode;
@Bean
public DefaultMQPushConsumer getRocketMQConsumer() throws MQClientException {
// ConsumerMode consumerMode = new ConsumerMode();
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerMode.getGroupName());
consumer.setNamesrvAddr(consumerMode.getNamesrvAddr());
consumer.setConsumeThreadMin(consumerMode.getConsumeThreadMin());
consumer.setConsumeThreadMax(consumerMode.getConsumeThreadMax());
consumer.registerMessageListener(new RocketMsgListener());
/**
* 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
* 2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
* 3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费
* 以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始
*/
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
/**
* CLUSTERING (集群模式) ：默认模式，同一个ConsumerGroup(groupName相同)每个consumer只消费所订阅消息的一部分内容，同一个ConsumerGroup里所有的Consumer消息加起来才是所
* 订阅topic整体，从而达到负载均衡的目的
* BROADCASTING (广播模式) ：同一个ConsumerGroup每个consumer都消费到所订阅topic所有消息，也就是一个消费会被多次分发，被多个consumer消费。
*
*/
// consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.setVipChannelEnabled(false);
consumer.setConsumeMessageBatchMaxSize(consumerMode.getConsumeMessageBatchMaxSize());
try {
/**
* 订阅topic，可以对指定消息进行过滤，例如："TopicTest","tagl||tag2||tag3",*或null表示topic所有消息
*/
consumer.subscribe(MessageCodeEnum.ORDER_MESSAGE.getCode(),"*");
consumer.subscribe(MessageCodeEnum.USER_MESSAGE.getCode(),"*");
consumer.start();
log.info("消费者初始化成功：{}", consumer.toString());
} catch (MQClientException e) {
e.printStackTrace();
log.error("消费者初始化失败：{}",e.getMessage());
}
return consumer;
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/config/MessageConfig.java
-27
View File
@@ -1,27 +0,0 @@
package com.ruoyi.testrocketmq.config;
/**
* @author yz
*/
public class MessageConfig {
private Class<?> messageClass;
private boolean orderlyMessage;
public Class<?> getMessageClass() {
return messageClass;
}
public void setMessageClass(Class<?> messageClass) {
this.messageClass = messageClass;
}
public boolean isOrderlyMessage() {
return orderlyMessage;
}
public void setOrderlyMessage(boolean orderlyMessage) {
this.orderlyMessage = orderlyMessage;
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/config/ProducerConfig.java
-48
View File
@@ -1,48 +0,0 @@
package com.ruoyi.testrocketmq.config;
import com.ruoyi.testrocketmq.model.ProducerMode;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@Slf4j
public class ProducerConfig {
public static DefaultMQProducer producer;
@Autowired
private ProducerMode producerMode;
@Bean
public DefaultMQProducer getRocketMQProducer() {
producer = new DefaultMQProducer(producerMode.getGroupName());
producer.setNamesrvAddr(producerMode.getNamesrvAddr());
//如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
if(producerMode.getMaxMessageSize()!=null){
producer.setMaxMessageSize(producerMode.getMaxMessageSize());
}
if(producerMode.getSendMsgTimeout()!=null){
producer.setSendMsgTimeout(producerMode.getSendMsgTimeout());
}
//如果发送消息失败，设置重试次数，默认为2次
if(producerMode.getRetryTimesWhenSendFailed()!=null){
producer.setRetryTimesWhenSendFailed(producerMode.getRetryTimesWhenSendFailed());
}
producer.setVipChannelEnabled(false);
try {
producer.start();
log.info("生产者初始化成功：{}",producer.toString());
} catch (MQClientException e) {
log.error("生产者初始化失败：{}",e.getMessage());
}
return producer;
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/consumer/RocketMsgListener.java
-100
View File
@@ -1,100 +0,0 @@
package com.ruoyi.testrocketmq.consumer;
import com.ruoyi.testrocketmq.enums.MessageCodeEnum;
import com.ruoyi.testrocketmq.producer.ConsumeException;
import lombok.extern.slf4j.Slf4j;
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.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.io.UnsupportedEncodingException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* 消息监听
*/
@Slf4j
@Component
public class RocketMsgListener implements MessageListenerConcurrently {
/**
* 消费消息
*
* @param list msgs.size() >= 1
* DefaultMQPushConsumer.consumeMessageBatchMaxSize=1，you can modify here
* 这里只设置为1，当设置为多个时，list中只要有一条消息消费失败，就会整体重试
* @param consumeConcurrentlyContext 上下文信息
* @return 消费状态 成功（CONSUME_SUCCESS）或者 重试 (RECONSUME_LATER)
*/
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
if (!CollectionUtils.isEmpty(list)) {
for (MessageExt messageExt : list) {
// 消息内容
String body = new String(messageExt.getBody());
log.info("接受到的消息为：{}", body);
String tags = messageExt.getTags();
String topic = messageExt.getTopic();
String msgId = messageExt.getMsgId();
String keys = messageExt.getKeys();
int reConsume = messageExt.getReconsumeTimes();
// 消息已经重试了3次，如果不需要再次消费，则返回成功
if (reConsume == 3) {
// TODO 补偿信息
//smsLogService.insertLog(topic, tags, msgId, keys, body, "【" + EnumUtil.getStrMsgByCode(tags, TagsCodeEnum.class) + "】消费失败");
log.error("消息重试超过3次，消费失败！");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
// 订单超时处理
if (MessageCodeEnum.ORDER_MESSAGE.getCode().equals(topic)) {
if (MessageCodeEnum.ORDER_TIMEOUT_TAG.getCode().equals(tags)) {
// //获取订单
// DealUserOrder dealUserOrder = pcRemoteDealUserOrderService.selectDealUserOrderByOrderNumber(keys);
// if (dealUserOrder != null) {
// //订单状态超时未支付关闭订单 处理
// if (dealUserOrder.getStatus().equals("1")) {
// DealUserOrder dealUserOrders = new DealUserOrder();
// dealUserOrders.setOrderId(dealUserOrder.getOrderId());
// dealUserOrders.setStatus("4");
// pcRemoteDealUserOrderService.updateDealUserOrder(dealUserOrders);
// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
// }
// log.info("Order does not exist.");
// }
log.info("Consumption success：" + body);
DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
log.info("Consumption time：{}", format.format(new Date()));
} else {
log.info("未匹配到Tag【{}】" + tags);
}
}
}
}
// 消息消费成功
//ConsumeConcurrentlyStatus.RECONSUME_LATER broker会根据设置的messageDelayLevel发起重试，默认16次
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
/**
* 异常处理
*
* @param e 捕获的异常
* @return 消息消费结果
*/
private static ConsumeConcurrentlyStatus handleException(final Exception e) {
Class exceptionClass = e.getClass();
if (exceptionClass.equals(UnsupportedEncodingException.class)) {
log.error(e.getMessage());
} else if (exceptionClass.equals(ConsumeException.class)) {
log.error(e.getMessage());
}
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/enums/MessageCodeEnum.java
-60
View File
@@ -1,60 +0,0 @@
package com.ruoyi.testrocketmq.enums;
import lombok.Getter;
@Getter
public enum MessageCodeEnum {
/**
* 消息模块主题
*/
MESSAGE_TOPIC("elink-message","消息服务模块topic名称"),
/**
* 系统消息
*/
NOTE_MESSAGE("system-message","系统消息服务模块topic名称"),
/**
* 用户消息
*/
USER_MESSAGE("user-message","用户消息服务模块topic名称"),
/**
* 订单消息
*/
ORDER_MESSAGE("order-message","订单消息服务模块topic名称"),
/**
* 平台编号
*/
USER_MESSAGE_TAG("user_message_tag","用户消息推送"),
NOTE_MESSAGE_TAG("system_message_tag","系统消息推送"),
ORDER_MESSAGE_TAG("order_message_tag","订单消息推送"),
/**
* 订单处理编号
*/
//订单超时处理
ORDER_TIMEOUT_TAG("order_timeout_tag","订单超时处理");
private final String code;
private final String msg;
MessageCodeEnum(String code, String msg){
this.code = code;
this.msg = msg;
}
public static String valuesOfType(String code) {
String value = "";
for (MessageCodeEnum e : MessageCodeEnum.values()) {
if (code.equals(e.code)) {
value = e.msg;
}
}
return value;
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/model/ConsumerMode.java
-22
View File
@@ -1,22 +0,0 @@
package com.ruoyi.testrocketmq.model;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
@Data
@Configuration
@Component
public class ConsumerMode {
@Value("${suning.rocketmq.namesrvAddr}")
private String namesrvAddr;
@Value("${suning.rocketmq.conumer.groupName}")
private String groupName ;
@Value("${suning.rocketmq.conumer.consumeThreadMin}")
private int consumeThreadMin;
@Value("${suning.rocketmq.conumer.consumeThreadMax}")
private int consumeThreadMax;
@Value("${suning.rocketmq.conumer.consumeMessageBatchMaxSize}")
private int consumeMessageBatchMaxSize;
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/model/ProducerMode.java
-25
View File
@@ -1,25 +0,0 @@
package com.ruoyi.testrocketmq.model;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Configuration;
/**
* 生产者初始化
*/
@RefreshScope
@Data
@Configuration
public class ProducerMode {
@Value("${suning.rocketmq.producer.groupName}")
private String groupName;
@Value("${suning.rocketmq.namesrvAddr}")
private String namesrvAddr;
@Value("${suning.rocketmq.producer.maxMessageSize}")
private Integer maxMessageSize;
@Value("${suning.rocketmq.producer.sendMsgTimeout}")
private Integer sendMsgTimeout;
@Value("${suning.rocketmq.producer.retryTimesWhenSendFailed}")
private Integer retryTimesWhenSendFailed;
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/producer/AsyncProducer.java
-47
View File
@@ -1,47 +0,0 @@
package com.ruoyi.testrocketmq.producer;
import com.ruoyi.testrocketmq.config.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
public class AsyncProducer {
@Autowired
private ProducerConfig producerConfig;
/**
* 发送异步的消息
* @param topic 主题
* @param tag 标签
* @param key 自定义的key，根据业务来定
* @param value 消息的内容
* @return org.apache.rocketmq.client.producer.SendResult
*/
// public SendResult sendAsyncProducerMessage(String topic, String tag, String key, String value) throws UnsupportedEncodingException {
//
// try {
// DefaultMQProducer defaultMQProducer = producerConfig.producer;
// //Create a message instance, specifying topic, tag and message body.
// Message msg = new Message(topic, tag, key,value.getBytes(RemotingHelper.DEFAULT_CHARSET));
// defaultMQProducer.send(msg, new SendCallback() {
// // 异步回调的处理
// @Override
// public void onSuccess(SendResult sendResult) {
// System.out.printf("%-10d 异步发送消息成功 %s %n", msg, sendResult.getMsgId());
// }
//
// @Override
// public void onException(Throwable e) {
// System.out.printf("%-10d 异步发送消息失败 %s %n", msg, e);
// e.printStackTrace();
// }
// });
// } catch (MQClientException e) {
// e.printStackTrace();
// } catch (RemotingException e) {
// e.printStackTrace();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// return null;
// }
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/producer/ConsumeException.java
-20
View File
@@ -1,20 +0,0 @@
package com.ruoyi.testrocketmq.producer;
/**
* @author 影子
*/
public class ConsumeException extends RuntimeException{
private static final long serialVersionUID = 4093867789628938836L;
public ConsumeException(String message) {
super(message);
}
public ConsumeException(Throwable cause) {
super(cause);
}
public ConsumeException(String message, Throwable cause) {
super(message, cause);
}
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/producer/MessageContext.java
-34
View File
@@ -1,34 +0,0 @@
package com.ruoyi.testrocketmq.producer;
import lombok.Data;
import lombok.ToString;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
/**
* 消费时，当前所消费的消息的上下文信息
*
* @author jolly
*/
@ToString
@Data
public final class MessageContext {
/**
* 所消费消息所在的消息队列
*
* @see MessageQueue
*/
private MessageQueue messageQueue;
/**
* 所消费的消息的扩展属性
*
* @see MessageExt
*/
private MessageExt messageExt;
}
ruoyi-rocketmq/src/test/java/com/ruoyi/testrocketmq/producer/MessageProducer.java
-110
View File
@@ -1,110 +0,0 @@
package com.ruoyi.testrocketmq.producer;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.io.UnsupportedEncodingException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import static com.ruoyi.rocketmq.config.ProducerConfig.producer;
/**
* 消息发送
*/
@Slf4j
public class MessageProducer {
/**
* 同步发送消息
* @param topic 主题
* @param tag 标签
* @param key 自定义的key，根据业务来定
* @param value 消息的内容
* @return org.apache.rocketmq.client.producer.SendResult
*/
public SendResult sendSynchronizeMessage(String topic, String tag, String key, String value){
String body = "topic：【"+topic+"】, tag：【"+tag+"】, key：【"+key+"】, value：【"+value+"】";
try {
Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET));
System.out.println("生产者发送消息:"+ JSON.toJSONString(value));
SendResult result = producer.send(msg);
return result;
} catch (UnsupportedEncodingException e) {
log.error("消息初始化失败！body：{}",body);
} catch (MQClientException | InterruptedException | RemotingException | MQBrokerException e) {
log.error("消息发送失败! body：{}",body);
}
return null;
}
/**
* 发送有序的消息
* @param messagesList Message集合
* @param messageQueueNumber 消息队列编号
* @return org.apache.rocketmq.client.producer.SendResult
*/
public SendResult sendOrderlyMessage(List<Message> messagesList, int messageQueueNumber) {
SendResult result = null;
for (Message message : messagesList) {
try {
// DefaultMQProducer defaultMQProducer = ProducerConfig.producer.send(message);
// System.out.println(defaultMQProducer);
result = producer.send(message, (list, msg, arg) -> {
Integer queueNumber = (Integer) arg;
return list.get(queueNumber);
}, messageQueueNumber);
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
log.error("发送有序消息失败");
return result;
}
}
return result;
}
/**
* 推送延迟消息
* @param topic
* @param tag
* @param key
* @return boolean
*/
public SendResult sendDelayMessage(String topic, String tag, String key, String value)
{
SendResult result = null;
try
{
Message msg = new Message(topic,tag,key, value.getBytes(RemotingHelper.DEFAULT_CHARSET));
//设置消息延迟级别，我这里设置5，对应就是延时一分钟
// "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
msg.setDelayTimeLevel(4);
// 发送消息到一个Broker
result = producer.send(msg);
// 通过sendResult返回消息是否成功送达
log.info("发送延迟消息结果：======sendResult：{}", result);
DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
log.info("发送时间：{}", format.format(new Date()));
return result;
}
catch (Exception e)
{
e.printStackTrace();
log.error("延迟消息队列推送消息异常:{},推送内容:{}", e.getMessage(), result);
}
return result;
}
}