Skip to content

Y
yd-notice
Commit 9f1becff by zhangxingmin
Options

push

parent 7aeea4f7
Showing with 293 additions and 9 deletions
yd-notice-api/src/main/java/com/yd/notice/api/service/impl/ApiNotificationTaskServiceImpl.java
...@@ -29,7 +29,8 @@ public class ApiNotificationTaskServiceImpl implements ApiNotificationTaskServic ...@@ -29,7 +29,8 @@ public class ApiNotificationTaskServiceImpl implements ApiNotificationTaskServic
request.getChannelBizId(), request.getChannelBizId(),
request.getTemplateBizId(), request.getTemplateBizId(),
request.getReceiver(), request.getReceiver(),
request.getParams() request.getParams(),
request.getReceiverType()
); );
ApiSendResponse response = new ApiSendResponse(); ApiSendResponse response = new ApiSendResponse();
response.setTaskBizId(taskBizId); response.setTaskBizId(taskBizId);
......
yd-notice-feign/src/main/java/com/yd/notice/feign/request/ApiSendRequest.java
...@@ -17,7 +17,7 @@ public class ApiSendRequest { ...@@ -17,7 +17,7 @@ public class ApiSendRequest {
private String templateBizId; private String templateBizId;
/** /**
* 消息接收人标识 * 单个消息接收人标识
* 具体格式由渠道和接收人类型决定： * 具体格式由渠道和接收人类型决定：
* - 企业微信：userid（多个用 | 分隔，@all 表示全员） * - 企业微信：userid（多个用 | 分隔，@all 表示全员）
* - 短信：手机号 * - 短信：手机号
...@@ -26,6 +26,11 @@ public class ApiSendRequest { ...@@ -26,6 +26,11 @@ public class ApiSendRequest {
private String receiver; private String receiver;
/** /**
* 单个接收人类型：sys/userid/mobile/email/openid
*/
private String receiverType;
/**
* 模板参数，JSON 字符串格式，用于替换模板中的占位符 * 模板参数，JSON 字符串格式，用于替换模板中的占位符
* 示例：{"name":"张三","orderId":"123456"} * 示例：{"name":"张三","orderId":"123456"}
* 占位符格式：{{name}}、{{orderId}} * 占位符格式：{{name}}、{{orderId}}
......
yd-notice-service/src/main/java/com/yd/notice/service/config/RestTemplateConfig.java 0 → 100644
package com.yd.notice.service.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.client.RestTemplate;
@Configuration
public class RestTemplateConfig {
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
\ No newline at end of file
yd-notice-service/src/main/java/com/yd/notice/service/model/NotificationTemplate.java
...@@ -91,6 +91,12 @@ public class NotificationTemplate implements Serializable { ...@@ -91,6 +91,12 @@ public class NotificationTemplate implements Serializable {
private Integer status; private Integer status;
/** /**
* 最大重试次数
*/
@TableField("max_retry")
private Integer maxRetry;
/**
* 排序 * 排序
*/ */
@TableField("sort") @TableField("sort")
......
yd-notice-service/src/main/java/com/yd/notice/service/send/WechatMpMessageSender.java 0 → 100644
package com.yd.notice.service.send;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yd.notice.service.model.ChannelConfig;
import com.yd.notice.service.model.NotificationTask;
import com.yd.notice.service.model.NotificationTemplate;
import com.yd.notice.service.service.IChannelConfigService;
import com.yd.notice.service.service.INotificationTemplateService;
import com.yd.notice.service.utils.WechatMpTokenUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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.RestTemplate;
import java.util.HashMap;
import java.util.Map;
/**
* 微信公众号模板消息发送器
* 渠道类型: wechat_mp
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class WechatMpMessageSender implements MessageSender {
private static final String SEND_URL = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=%s";
private final IChannelConfigService channelConfigService;
private final INotificationTemplateService templateService;
private final WechatMpTokenUtil wechatMpTokenUtil;
private final RestTemplate restTemplate;
/**
* 不可重试的错误码（永久失败）
*/
private static final String[] NON_RETRYABLE_CODES = {
"40003", // 不合法的 OpenID[reference:3]
"40037", // template_id 不正确[reference:4]
"40014", // 不合法的 access_token
"40127", // 模板消息已下架
"41001", // 缺少 access_token 参数
"41002", // 缺少 appid 参数
"41003", // 缺少 refresh_token 参数
"41004", // 缺少 secret 参数
"41005", // 缺少多媒体文件数据
"41006", // 缺少 media_id 参数
"41007", // 缺少子菜单数据
"41008", // 缺少 OAuth code
"41009", // 缺少 openid
"43004", // 需要关注公众号才能接收消息
"43101", // 用户拒绝接受消息（用户取消订阅或未关注）[reference:5]
"45009", // 接口调用超过限制
"46001", // 媒体文件不存在
"46002", // 媒体文件不合法
"46003", // 媒体文件超过限制
"47001", // 解析 JSON/XML 内容错误
};
@Override
public SendResult send(NotificationTask task) {
log.info("进入微信公众号模板消息发送器, taskBizId={}", task.getTaskBizId());
try {
// 1. 获取渠道配置
ChannelConfig config = channelConfigService.getByChannelBizId(task.getChannelBizId());
log.info("微信公众号模板消息发送器=>获取渠道配置:{}",JSON.toJSONString(config));
if (config == null || config.getStatus() != 1) {
return SendResult.failNonRetryable("CONFIG_NOT_EXIST", "渠道配置不存在或已禁用");
}
JSONObject wxConfig = JSONObject.parseObject(config.getConfigValue());
log.info("微信公众号模板消息发送器=>配置值:{}",JSON.toJSONString(wxConfig));
String appid = wxConfig.getString("appid");
String secret = wxConfig.getString("secret");
if (appid == null || secret == null) {
return SendResult.failNonRetryable("CONFIG_INVALID", "渠道配置缺少 appid 或 secret");
}
// 2. 获取模板配置
NotificationTemplate template = templateService.getByTemplateBizId(task.getTemplateBizId());
log.info("微信公众号模板消息发送器=>获取模板配置:{}",JSON.toJSONString(template));
if (template == null) {
return SendResult.failNonRetryable("TEMPLATE_NOT_EXIST", "模板不存在");
}
// extra_template 字段存储微信侧的模板 ID
String wxTemplateId = template.getExtraTemplate();
if (wxTemplateId == null || wxTemplateId.isEmpty()) {
return SendResult.failNonRetryable("TEMPLATE_ID_EMPTY", "未配置微信模板ID");
}
// 3. 获取 access_token
String accessToken = wechatMpTokenUtil.getAccessToken(appid, secret);
log.info("微信公众号模板消息发送器=>获取 access_token:{}",accessToken);
// 4. 构建请求体
// task.getContent() 已经是渲染后的 JSON 字符串
// 格式: {"keyword1":{"value":"xxx"},"keyword2":{"value":"yyy"}}
JSONObject dataJson = JSONObject.parseObject(task.getContent());
log.info("微信公众号模板消息发送器=>构建请求体:{}",JSON.toJSONString(dataJson));
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("touser", task.getReceiver()); // 公众号 openid
requestBody.put("template_id", wxTemplateId);
requestBody.put("data", dataJson);
// 可选：如果模板消息需要跳转小程序（需公众号和小程序关联）
// 可在 NotificationTemplate 的 extra_params 中配置 miniprogram 信息
// 或从 task.getExtraParams() 中获取
// requestBody.put("miniprogram", miniprogramMap);
log.info("微信公众号模板消息请求参数: {}", JSON.toJSONString(requestBody));
// 5. 发送请求
String url = String.format(SEND_URL, accessToken);
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<String> entity = new HttpEntity<>(JSON.toJSONString(requestBody), headers);
String response = restTemplate.postForObject(url, entity, String.class);
JSONObject result = JSONObject.parseObject(response);
log.info("微信公众号模板消息发送器=>发送请求响应返回:{}",JSON.toJSONString(result));
Integer errCode = result.getInteger("errcode");
String errMsg = result.getString("errmsg");
log.info("微信公众号模板消息响应: errcode={}, errmsg={}", errCode, errMsg);
// 6. 处理结果
if (errCode == null || errCode == 0) {
log.info("微信公众号模板消息发送成功, taskBizId={}, openid={}", task.getTaskBizId(), task.getReceiver());
return SendResult.success();
} else {
String errorCode = String.valueOf(errCode);
if (isNonRetryable(errorCode)) {
log.warn("微信公众号模板消息发送失败（不可重试）, taskBizId={}, errCode={}, errMsg={}",
task.getTaskBizId(), errorCode, errMsg);
return SendResult.failNonRetryable(errorCode, errMsg);
}
log.warn("微信公众号模板消息发送失败（可重试）, taskBizId={}, errCode={}, errMsg={}",
task.getTaskBizId(), errorCode, errMsg);
return SendResult.failRetryable(errorCode, errMsg);
}
} catch (Exception e) {
log.error("微信公众号模板消息发送异常, taskBizId={}", task.getTaskBizId(), e);
return SendResult.failRetryable("SEND_EXCEPTION", e.getMessage());
}
}
/**
* 判断是否为不可重试错误
*/
private boolean isNonRetryable(String errorCode) {
for (String code : NON_RETRYABLE_CODES) {
if (code.equals(errorCode)) {
return true;
}
}
return false;
}
@Override
public String getSupportedChannelType() {
return "wechat_mp";
}
}
\ No newline at end of file
yd-notice-service/src/main/java/com/yd/notice/service/service/INotificationTaskService.java
...@@ -16,5 +16,5 @@ public interface INotificationTaskService extends IService<NotificationTask> { ...@@ -16,5 +16,5 @@ public interface INotificationTaskService extends IService<NotificationTask> {
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
String createAndSendTask(String channelBizId, String templateBizId, String createAndSendTask(String channelBizId, String templateBizId,
String receiver, String params); String receiver, String params, String receiverType);
} }
yd-notice-service/src/main/java/com/yd/notice/service/service/impl/NotificationTaskServiceImpl.java
...@@ -52,7 +52,8 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap ...@@ -52,7 +52,8 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap
@Override @Override
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
public String createAndSendTask(String channelBizId, String templateBizId, String receiver, String params) { public String createAndSendTask(String channelBizId, String templateBizId,
String receiver, String params, String receiverType) {
// 1. 查询模板 // 1. 查询模板
NotificationTemplate template = templateService.getOne( NotificationTemplate template = templateService.getOne(
new LambdaQueryWrapper<NotificationTemplate>() new LambdaQueryWrapper<NotificationTemplate>()
...@@ -79,10 +80,10 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap ...@@ -79,10 +80,10 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap
task.setTitle(title); task.setTitle(title);
task.setContent(content); task.setContent(content);
task.setReceiver(receiver); task.setReceiver(receiver);
task.setReceiverType("userid"); // 可根据实际动态设置 task.setReceiverType(receiverType); // 可根据实际动态设置
task.setStatus(0); // 待发送 task.setStatus(0); // 待发送
task.setRetryCount(0); task.setRetryCount(0);
task.setMaxRetry(3); task.setMaxRetry(template.getMaxRetry());//最大重试次数
task.setIsTiming(0); task.setIsTiming(0);
task.setCreateTime(LocalDateTime.now()); task.setCreateTime(LocalDateTime.now());
task.setUpdateTime(LocalDateTime.now()); task.setUpdateTime(LocalDateTime.now());
...@@ -127,7 +128,9 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap ...@@ -127,7 +128,9 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap
private boolean sendWithRetryAndRecord(NotificationTask task) { private boolean sendWithRetryAndRecord(NotificationTask task) {
log.info("带重试的发送sendWithRetryAndRecord=>入参NotificationTask:{}", JSON.toJSONString(task)); log.info("带重试的发送sendWithRetryAndRecord=>入参NotificationTask:{}", JSON.toJSONString(task));
//最大重试次数
int maxRetry = task.getMaxRetry(); int maxRetry = task.getMaxRetry();
//当前重试次数
int currentRetry = 0; int currentRetry = 0;
long waitMillis = 1000; long waitMillis = 1000;
...@@ -138,13 +141,16 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap ...@@ -138,13 +141,16 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap
log.error("渠道配置不存在，taskBizId: {}, channelBizId: {}", task.getTaskBizId(), task.getChannelBizId()); log.error("渠道配置不存在，taskBizId: {}, channelBizId: {}", task.getTaskBizId(), task.getChannelBizId());
return false; return false;
} }
//渠道类型
String channelType = channelConfig.getChannelType(); String channelType = channelConfig.getChannelType();
//根据渠道类型获取不同的消息发送器类
MessageSender sender = senderRouter.getSender(channelType); MessageSender sender = senderRouter.getSender(channelType);
log.info("获取发送类=>出参MessageSender:{}", JSON.toJSONString(sender)); log.info("获取发送类=>出参MessageSender:{}", JSON.toJSONString(sender));
log.info("sendWithRetryAndRecord=>参数currentRetry:{}", currentRetry); log.info("sendWithRetryAndRecord=>参数currentRetry:{}", currentRetry);
log.info("sendWithRetryAndRecord=>参数maxRetry:{}", maxRetry); log.info("sendWithRetryAndRecord=>参数maxRetry:{}", maxRetry);
while (currentRetry <= maxRetry) { while (currentRetry <= maxRetry) {
//当前重试次数小于等于最大重试次数，一直重试
// 创建发送记录（待发送） // 创建发送记录（待发送）
NotificationRecord record = buildInitialRecord(task, currentRetry); NotificationRecord record = buildInitialRecord(task, currentRetry);
log.info("进入while循环=>创建发送记录（待发送）:{}", JSON.toJSONString(record)); log.info("进入while循环=>创建发送记录（待发送）:{}", JSON.toJSONString(record));
...@@ -162,43 +168,59 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap ...@@ -162,43 +168,59 @@ public class NotificationTaskServiceImpl extends ServiceImpl<NotificationTaskMap
// 调用具体发送器 // 调用具体发送器
log.info("进入while循环=>调用具体发送器:{}", JSON.toJSONString(task)); log.info("进入while循环=>调用具体发送器:{}", JSON.toJSONString(task));
//调用具体消息发送器类
result = sender.send(task); result = sender.send(task);
log.info("进入while循环=>调用具体发送器出参返回:{}", JSON.toJSONString(result)); log.info("进入while循环=>调用具体发送器出参返回:{}", JSON.toJSONString(result));
//耗时（毫秒）
costTime = (int) (LocalDateTimeUtil.between(startTime, LocalDateTime.now()).toMillis()); costTime = (int) (LocalDateTimeUtil.between(startTime, LocalDateTime.now()).toMillis());
} catch (Exception e) { } catch (Exception e) {
//发送异常
log.info("发送异常，taskBizId: {}, retry: {}", task.getTaskBizId(), currentRetry, e); log.info("发送异常，taskBizId: {}, retry: {}", task.getTaskBizId(), currentRetry, e);
result = SendResult.failRetryable("UNKNOWN_ERROR", e.getMessage()); result = SendResult.failRetryable("UNKNOWN_ERROR", e.getMessage());
//耗时（毫秒）
costTime = (int) (LocalDateTimeUtil.between(startTime, LocalDateTime.now()).toMillis()); costTime = (int) (LocalDateTimeUtil.between(startTime, LocalDateTime.now()).toMillis());
} }
// 更新记录最终状态 //更新记录最终状态
//结果：是否成功 => 是:2 否:3
record.setStatus(result.isSuccess() ? 2 : 3); record.setStatus(result.isSuccess() ? 2 : 3);
//错误码
record.setErrorCode(result.getErrorCode()); record.setErrorCode(result.getErrorCode());
//错误信息
record.setErrorMsg(result.getErrorMsg()); record.setErrorMsg(result.getErrorMsg());
//耗时（毫秒）
record.setCostTime(costTime); record.setCostTime(costTime);
//发送时间
record.setSendTime(LocalDateTime.now()); record.setSendTime(LocalDateTime.now());
record.setUpdateTime(LocalDateTime.now()); record.setUpdateTime(LocalDateTime.now());
recordService.updateById(record); recordService.updateById(record);
//判断是否成功，成功就直接结束循环，直接结束
if (result.isSuccess()) { if (result.isSuccess()) {
return true; return true;
} }
// 失败处理 // 失败处理 => 当前重试次数+1
currentRetry++; currentRetry++;
task.setRetryCount(currentRetry); task.setRetryCount(currentRetry);
task.setUpdateTime(LocalDateTime.now()); task.setUpdateTime(LocalDateTime.now());
taskMapper.updateById(task); taskMapper.updateById(task);
if (currentRetry <= maxRetry) { if (currentRetry <= maxRetry) {
//根据返回结果的是否重试状态，来决定是否重试。
if (!result.isRetryable()) { if (!result.isRetryable()) {
log.info("检测到不可重试错误，放弃重试，taskBizId: {}, errorCode: {}", task.getTaskBizId(), result.getErrorCode()); log.info("检测到不可重试错误，放弃重试，taskBizId: {}, errorCode: {}", task.getTaskBizId(), result.getErrorCode());
break; break;
} }
log.info("发送失败，{}秒后进行第{}次重试，taskBizId: {}", waitMillis / 1000, currentRetry, task.getTaskBizId()); log.info("发送失败，{}秒后进行第{}次重试，taskBizId: {}", waitMillis / 1000, currentRetry, task.getTaskBizId());
try { try {
//睡眠一定毫秒再次重试
Thread.sleep(waitMillis); Thread.sleep(waitMillis);
//每次重试前，先等一会儿；并且等待的时间越来越长，但最长不超过30秒。
// waitMillis = Math.min(waitMillis * 2, 30000);
//这行是计算下一次重试时要等多长时间：
//先把当前等待时间 乘以 2（指数增长）。
//然后用 Math.min(..., 30000) 取最小值，确保翻倍后的时间绝不超过 30000 毫秒（即 30 秒）。
waitMillis = Math.min(waitMillis * 2, 30000); waitMillis = Math.min(waitMillis * 2, 30000);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
......
yd-notice-service/src/main/java/com/yd/notice/service/utils/WechatMpTokenUtil.java 0 → 100644
package com.yd.notice.service.utils;
import com.alibaba.fastjson.JSONObject;
import com.yd.common.utils.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
@Slf4j
@Component
public class WechatMpTokenUtil {
private static final String TOKEN_URL = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s";
private static final String REDIS_KEY_PREFIX = "wechat_mp:access_token:";
@Resource
private RedisUtil redisUtil;
@Resource
private RestTemplate restTemplate;
/**
* 获取公众号 access_token（带 Redis 缓存）
*/
public String getAccessToken(String appid, String secret) {
String redisKey = REDIS_KEY_PREFIX + appid;
String cachedToken = redisUtil.getCacheObject(redisKey);
if (cachedToken != null) {
log.debug("从缓存获取 access_token: {}", cachedToken);
return cachedToken;
}
String url = String.format(TOKEN_URL, appid, secret);
try {
String response = restTemplate.getForObject(url, String.class);
JSONObject json = JSONObject.parseObject(response);
String accessToken = json.getString("access_token");
Integer expiresIn = json.getInteger("expires_in");
if (accessToken == null || accessToken.isEmpty()) {
String errMsg = json.getString("errmsg");
throw new RuntimeException("获取 access_token 失败: " + errMsg);
}
// 缓存 7000 秒（略短于 7200 秒，留有余量）
int cacheSeconds = (expiresIn != null && expiresIn > 0) ? expiresIn - 200 : 7000;
redisUtil.setCacheObject(redisKey, accessToken, cacheSeconds, TimeUnit.SECONDS);
log.info("获取并缓存 access_token 成功，有效期 {} 秒", cacheSeconds);
return accessToken;
} catch (Exception e) {
log.error("获取 access_token 异常", e);
throw new RuntimeException("获取微信公众号 access_token 失败", e);
}
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment