msb_345200
/
beacon-cloud
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: msb_345200:9b8566c1aad1e1d366d2af5991cda85e188b9dbb
Branches Tags
msb_345200:main
..
compare: msb_345200:b78c0054769a3ed3a54f632004bd2ded399d3af5
Branches Tags
msb_345200:main

No commits in common. '9b8566c1aad1e1d366d2af5991cda85e188b9dbb' and 'b78c0054769a3ed3a54f632004bd2ded399d3af5' have entirely different histories.
9b8566c1aa ... b78c005476

11 changed files with 5 additions and 453 deletions
Show Stats

8
beacon-common/src/main/java/com/mashibing/common/pojo/StandardReport.java
Unescape View File

@ -1,9 +1,5 @@
package com.mashibing.common.pojo; package com.mashibing.common.pojo;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;
import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;
import com.mashibing.common.annotation.Description; import com.mashibing.common.annotation.Description;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Data; import lombok.Data;
@ -36,8 +32,6 @@ public class StandardReport implements Serializable {
@Description("目标手机号，客户请求传递的") @Description("目标手机号，客户请求传递的")
private String mobile; private String mobile;
@JsonSerialize(using = LocalDateTimeSerializer.class)
@JsonDeserialize(using = LocalDateTimeDeserializer.class)
@Description("短信的发送时间，当前系统时间") @Description("短信的发送时间，当前系统时间")
private LocalDateTime sendTime; private LocalDateTime sendTime;
@ -53,7 +47,5 @@ public class StandardReport implements Serializable {
@Description("回调url") @Description("回调url")
private String callbackUrl; private String callbackUrl;
@Description("推送报告推送次数")
private Integer pushTimes = 0;
} }

130
beacon-common/src/main/java/com/mashibing/common/utils/JsonUtil.java
Unescape View File

@ -2,145 +2,23 @@ package com.mashibing.common.utils;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.SerializationFeature;
import java.util.Map;
/** /**
* @author heqijun * @author heqijun
* @ClassName: JsonUtil * @ClassName: JsonUtil
* @Description: JsonUtil工具类 * @Description: TODO(这里用一句话描述这个类的作用)
* @date 2025/6/12 18:46 * @date 2025/6/12 18:46
*/ */
public class JsonUtil { public class JsonUtil {
private static final ObjectMapper objectMapper = new ObjectMapper(); private static ObjectMapper objectMapper = new ObjectMapper();
static {
// 配置ObjectMapper
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
objectMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
}
/** public static String obj2Json(Object obj) {
* 将对象转换为JSON字符串
*
* @param obj 任意Java对象
* @return JSON字符串
*/
public static String objToJson(Object obj) {
try { try {
return objectMapper.writeValueAsString(obj); return objectMapper.writeValueAsString(obj);
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new RuntimeException("对象转JSON失败", e); throw new RuntimeException("转换JSON失败！！！");
}
}
/**
* 将JSON字符串转换为指定类型的对象
*
* @param json JSON字符串
* @param clazz 目标对象类型
* @return 转换后的对象
*/
public static <T> T jsonToObj(String json, Class<T> clazz) {
try {
return objectMapper.readValue(json, clazz);
} catch (JsonProcessingException e) {
throw new RuntimeException("JSON转对象失败", e);
}
}
/**
* 将JSON字符串转换为Map
*
* @param json JSON字符串
* @return Map对象
*/
public static Map<String, Object> jsonToMap(String json) {
try {
return objectMapper.readValue(json, new TypeReference<Map<String, Object>>() {});
} catch (JsonProcessingException e) {
throw new RuntimeException("JSON转Map失败", e);
}
}
/**
* 将Map转换为JSON字符串
*
* @param map Map对象
* @return JSON字符串
*/
public static String mapToJson(Map<String, Object> map) {
try {
return objectMapper.writeValueAsString(map);
} catch (JsonProcessingException e) {
throw new RuntimeException("Map转JSON失败", e);
}
}
/**
* 将Map转换为指定类型的对象
*
* @param map Map对象
* @param clazz 目标对象类型
* @return 转换后的对象
*/
public static <T> T mapToObj(Map<String, Object> map, Class<T> clazz) {
try {
String json = mapToJson(map);
return jsonToObj(json, clazz);
} catch (Exception e) {
throw new RuntimeException("Map转对象失败", e);
}
}
/**
* 将对象转换为Map
*
* @param obj 任意Java对象
* @return Map对象
*/
public static Map<String, Object> objToMap(Object obj) {
try {
String json = objToJson(obj);
return jsonToMap(json);
} catch (Exception e) {
throw new RuntimeException("对象转Map失败", e);
}
}
/**
* 通用转换方法，支持任意类型转换
*
* @param source 源数据
* @param targetType 目标类型
* @return 转换后的结果
*/
@SuppressWarnings("unchecked")
public static <T> T convert(Object source, Class<T> targetType) {
if (source == null) {
return null;
}
if (targetType.isAssignableFrom(source.getClass())) {
return (T) source;
}
if (source instanceof String) {
// JSON字符串转对象
return jsonToObj((String) source, targetType);
} else if (source instanceof Map) {
// Map转对象
return mapToObj((Map<String, Object>) source, targetType);
} else {
// 对象转JSON再转目标类型
String json = objToJson(source);
return jsonToObj(json, targetType);
} }
} }
} }

58
beacon-push/pom.xml
Unescape View File

@ -1,58 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.mashibing</groupId>
<artifactId>beacon-cloud</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<artifactId>beacon-push</artifactId>
<dependencies>
<!-- web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- nacos注册中心 -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
<!-- nacos配置中心 -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
<!-- AMQP -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 测试 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
<!-- lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!-- common公共组件 -->
<dependency>
<groupId>com.mashibing</groupId>
<artifactId>beacon-common</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
</dependencies>
</project>

22
beacon-push/src/main/java/com/mashibing/push/PushApplication.java
Unescape View File

@ -1,22 +0,0 @@
package com.mashibing.push;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;
/**
* @author heqijun
* @ClassName: PushApplication
* @Description: 推送模块启动类
* @date 2025/6/12 19:41
*/
@SpringBootApplication
@EnableFeignClients
@EnableDiscoveryClient
public class PushApplication {
public static void main(String[] args) {
SpringApplication.run(PushApplication.class, args);
}
}

46
beacon-push/src/main/java/com/mashibing/push/config/RabbitMQConfig.java
Unescape View File

@ -1,46 +0,0 @@
package com.mashibing.push.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* @author heqijun
* @ClassName: RabbitMQConfig
* @Description: 声明延迟交换机并绑定队列
* @date 2025/6/12 21:35
*/
@Configuration
public class RabbitMQConfig {
public static final String DELAYED_EXCHANGE = "push_delayed_exchange";
public static final String DELAYED_QUEUE = "push_delayed_queue";
private static final String DELAYED_EXCHANGE_TYPE = "x-delayed-message";
private static final String DELAYED_ROUTING_TYPE_KEY = "x-delayed-type";
private static final String DELAYED_ROUTING_TYPE_FANOUT = "fanout";
@Bean
public Exchange delayedExchange() {
Map<String, Object> args = new HashMap<String, Object>();
args.put(DELAYED_ROUTING_TYPE_KEY, DELAYED_ROUTING_TYPE_FANOUT);
return new CustomExchange(DELAYED_EXCHANGE, DELAYED_EXCHANGE_TYPE, true, false, args);
}
@Bean
public Queue delayedQueue() {
return QueueBuilder.durable(DELAYED_QUEUE).build();
}
@Bean
public Binding delayedExchangeBinding(Exchange delayedExchange, Queue delayedQueue) {
return BindingBuilder.bind(delayedQueue).to(delayedExchange).with("").noargs();
}
}

23
beacon-push/src/main/java/com/mashibing/push/config/RestTemplateConfig.java
Unescape View File

@ -1,23 +0,0 @@
package com.mashibing.push.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
/**
* @author heqijun
* @ClassName: RestTemplateConfig
* @Description: RestTemplateConfig配置类
* @date 2025/6/12 19:54
*/
@Slf4j
@Component
public class RestTemplateConfig {
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}

15
beacon-push/src/main/java/com/mashibing/push/feignclient/Cacheclient.java
Unescape View File

@ -1,15 +0,0 @@
package com.mashibing.push.feignclient;
import com.mashibing.common.clients.BeaconCacheClient;
import org.springframework.cloud.openfeign.FeignClient;
/**
* @author heqijun
* @ClassName: Cacheclient
* @Description: push模块调cache模块的feign client
* @date 2025/6/12 19:55
*/
@FeignClient("beacon-cache")
public interface Cacheclient extends BeaconCacheClient {
}

136
beacon-push/src/main/java/com/mashibing/push/mq/PushReportListener.java
Unescape View File

@ -1,136 +0,0 @@
package com.mashibing.push.mq;
import com.mashibing.common.constant.RabbitMQConstant;
import com.mashibing.common.pojo.StandardReport;
import com.mashibing.common.utils.JsonUtil;
import com.mashibing.push.config.RabbitMQConfig;
import com.mashibing.push.feignclient.Cacheclient;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import java.io.IOException;
import java.time.LocalDateTime;
/**
* @author heqijun
* @ClassName: PushReportListener
* @Description: 监听推送短信状态报告队列
* @date 2025/6/12 19:49
*/
@Slf4j
@Component
public class PushReportListener {
@Autowired
RestTemplate restTemplate;
@Autowired
Cacheclient cacheclient;
@Autowired
RabbitTemplate rabbitTemplate;
private static final String SUCCESS = "SUCCESS";
/**
* 重试时间间隔
*/
private static final int[] RETRY_INTERVAL_SECONDS = {0, 15000, 30000, 60000, 300000};
@RabbitListener(queues = {RabbitMQConstant.SMS_PUSH_REPORT})
public void consume(StandardReport report, Channel channel, Message message) throws IOException {
log.info("【推送模块】接收到状态报告消息: {}", report);
String callbackUrl = report.getCallbackUrl();
// 需要推送报告
if (report.getIsCallback() == 1L) {
if (StringUtils.isBlank(callbackUrl)) {
log.error("【推送模块】客户推送地址为空，无法推送！！！callbackUrl={}", callbackUrl);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
return;
}
// 推送报告
retry(report);
} else {
// 不需要推送报告
log.info("【推送模块】客户不需要推送报告！！！");
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
return;
}
ack(channel, message);
}
@RabbitListener(queues = {RabbitMQConfig.DELAYED_QUEUE})
public void delayConsume(StandardReport report, Channel channel, Message message) throws IOException {
retry(report);
//手动ack
ack(channel, message);
}
private static void ack(Channel channel, Message message) throws IOException {
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
log.info("【推送模块】消费消息完毕，手动ack！！！");
}
private boolean pushReport(StandardReport report) {
log.info("【推送模块】推送时间：{}", LocalDateTime.now());
boolean flag = false;
// 声明发送参数
String body = JsonUtil.objToJson(report);
// 声明restTemplate模板代码
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<Object> httpEntity = new HttpEntity<>(body, headers);
//发请求
try {
report.setPushTimes(report.getPushTimes() + 1);
String response = restTemplate.postForObject("http://" + report.getCallbackUrl(), httpEntity, String.class);
flag = SUCCESS.equals(response);
} catch (RestClientException e) {
log.error("【推送模块】推送消息到客户异常：errorMessage{}", e.getMessage());
// e.printStackTrace();
}
return flag;
}
private void retry(StandardReport report) {
// 推送报告
log.info("【推送模块】第{}次推送状态报告开始！！！report={}", report.getPushTimes() + 1, report);
boolean flag = pushReport(report);
if (!flag) {
// 重试
log.info("【推送模块】第{}次推送消息失败！！！report={}", report.getPushTimes(), report);
if (report.getPushTimes() >= 5) {
return;
}
// 放到延迟队列中
rabbitTemplate.convertAndSend(RabbitMQConfig.DELAYED_EXCHANGE, "", report, message1 -> {
//设置延迟时间
message1.getMessageProperties().setDelay(RETRY_INTERVAL_SECONDS[report.getPushTimes()]);
return message1;
});
} else {
log.info("【推送模块】第{}次推送消息成功！！！report={}", report.getPushTimes() + 1, report);
}
}
}

17
beacon-push/src/main/resources/bootstrap.yml
Unescape View File

@ -1,17 +0,0 @@
# 服务名称
spring:
application:
name: beacon-push
# 多环境
profiles:
active: dev
# nacos注册中心地址
cloud:
nacos:
discovery:
server-addr: 192.168.1.13:8848
# nacos配置中心地址:
config:
server-addr: 192.168.1.13:8848
file-extension: yml
# beacon-push-dev.yml

2
beacon-search/src/main/java/com/mashibing/search/mq/SmsWriteLogListener.java
Unescape View File

@ -36,7 +36,7 @@ public class SmsWriteLogListener {
log.info("接收到存储日志的信息，submit={}", submit); log.info("接收到存储日志的信息，submit={}", submit);
String sequenceId = submit.getSequenceId().toString(); String sequenceId = submit.getSequenceId().toString();
String json = JsonUtil.objToJson(submit); String json = JsonUtil.obj2Json(submit);
String year = LocalDateTime.now().getYear() + ""; String year = LocalDateTime.now().getYear() + "";
searchService.index(INDEX + year, sequenceId, json); searchService.index(INDEX + year, sequenceId, json);

1
pom.xml
Unescape View File

@ -15,7 +15,6 @@
<module>beacon-test</module> <module>beacon-test</module>
<module>beacon-strategy</module> <module>beacon-strategy</module>
<module>beacon-search</module> <module>beacon-search</module>
<module>beacon-push</module>
</modules> </modules>

Write Preview
Loading…
Cancel
Save