构建推送模块beacon-push + 实现推送状态报告 + 实现延迟队列推送五次

beacon-common/src/main/java/com/mashibing/common/pojo/StandardReport.java

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

beacon-common/src/main/java/com/mashibing/common/utils/JsonUtil.java

@@ -2,23 +2,145 @@ package com.mashibing.common.utils;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 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
  * @ClassName: JsonUtil
- * @Description: TODO(这里用一句话描述这个类的作用)
+ * @Description: JsonUtil工具类
  * @date 2025/6/12 18:46
  */
 
 public class JsonUtil {
 
-	private static ObjectMapper objectMapper = new ObjectMapper();
+	private static final 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 {
 			return objectMapper.writeValueAsString(obj);
 		} catch (JsonProcessingException e) {
-			throw new RuntimeException("转换JSON失败！！！");
+			throw new RuntimeException("对象转JSON失败", e);
+		}
+	}
+
+	/**
+	 * 将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);
 		}
 	}
 }

beacon-push/pom.xml

@@ -0,0 +1,58 @@
+<?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>

beacon-push/src/main/java/com/mashibing/push/PushApplication.java

@@ -0,0 +1,22 @@
+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);
+	}
+}

beacon-push/src/main/java/com/mashibing/push/config/RabbitMQConfig.java

@@ -0,0 +1,46 @@
+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();
+	}
+}

beacon-push/src/main/java/com/mashibing/push/config/RestTemplateConfig.java

@@ -0,0 +1,23 @@
+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();
+	}
+}

beacon-push/src/main/java/com/mashibing/push/feignclient/Cacheclient.java

@@ -0,0 +1,15 @@
+package com.mashibing.push.feignclient;
+
+import com.mashibing.common.clients.BeaconCacheClient;
+import org.springframework.cloud.openfeign.FeignClient;
+
+/**
+ * @author heqijun
+ * @ClassName: Cacheclient
+ * @Description: TODO(这里用一句话描述这个类的作用)
+ * @date 2025/6/12 19:55
+ */
+
+@FeignClient("beacon-cache")
+public interface Cacheclient extends BeaconCacheClient {
+}

beacon-push/src/main/java/com/mashibing/push/mq/PushReportListener.java

@@ -0,0 +1,136 @@
+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);
+		}
+	}
+
+
+}

beacon-push/src/main/resources/bootstrap.yml

@@ -0,0 +1,17 @@
+# 服务名称
+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

beacon-search/src/main/java/com/mashibing/search/mq/SmsWriteLogListener.java

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

pom.xml

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