本文目录导读：
我将为您提供一个完整的Java通知系统实现案例,包含多种通知渠道和灵活的设计模式。
核心接口设计
// 通知接口
public interface Notifier {
void send(Notification notification);
boolean supports(NotificationType type);
}
// 通知类型枚举
public enum NotificationType {
EMAIL, SMS, PUSH, WECHAT, SYSTEM
}
// 通知状态
public enum NotificationStatus {
PENDING, SENT, FAILED, RETRYING
}
通知实体类
// 通知类
public class Notification {
private String id;
private String title;
private String content;
private List<String> receivers;
private NotificationType type;
private NotificationStatus status;
private Map<String, Object> additionalData;
private LocalDateTime createTime;
private LocalDateTime sendTime;
private int retryCount;
// 使用建造者模式
public static class Builder {
private Notification notification = new Notification();
public Builder id(String id) {
notification.id = id;
return this;
}
public Builder title(String title) {
notification.title = title;
return this;
}
public Builder content(String content) {
notification.content = content;
return this;
}
public Builder receivers(List<String> receivers) {
notification.receivers = receivers;
return this;
}
public Builder type(NotificationType type) {
notification.type = type;
return this;
}
public Builder additionalData(Map<String, Object> data) {
notification.additionalData = data;
return this;
}
public Notification build() {
notification.createTime = LocalDateTime.now();
notification.status = NotificationStatus.PENDING;
notification.retryCount = 0;
return notification;
}
}
// Getters and Setters
// ... (省略具体实现)
}
具体通知实现
// 邮件通知
public class EmailNotifier implements Notifier {
private static final Logger logger = LoggerFactory.getLogger(EmailNotifier.class);
private JavaMailSender mailSender;
private TemplateEngine templateEngine;
public EmailNotifier(JavaMailSender mailSender, TemplateEngine templateEngine) {
this.mailSender = mailSender;
this.templateEngine = templateEngine;
}
@Override
public void send(Notification notification) {
try {
MimeMessage message = mailSender.createMimeMessage();
MimeMessageHelper helper = new MimeMessageHelper(message, true, "UTF-8");
helper.setTo(notification.getReceivers().toArray(new String[0]));
helper.setSubject(notification.getTitle());
helper.setText(buildEmailContent(notification), true);
mailSender.send(message);
notification.setStatus(NotificationStatus.SENT);
notification.setSendTime(LocalDateTime.now());
logger.info("邮件通知发送成功: {}", notification.getId());
} catch (Exception e) {
logger.error("邮件通知发送失败: {}", notification.getId(), e);
notification.setStatus(NotificationStatus.FAILED);
throw new NotificationException("邮件发送失败", e);
}
}
private String buildEmailContent(Notification notification) {
Context context = new Context();
context.setVariable("title", notification.getTitle());
context.setVariable("content", notification.getContent());
return templateEngine.process("notification-template", context);
}
@Override
public boolean supports(NotificationType type) {
return type == NotificationType.EMAIL;
}
}
// 短信通知
public class SmsNotifier implements Notifier {
private static final Logger logger = LoggerFactory.getLogger(SmsNotifier.class);
private SmsService smsService;
private String templateId;
public SmsNotifier(SmsService smsService, String templateId) {
this.smsService = smsService;
this.templateId = templateId;
}
@Override
public void send(Notification notification) {
try {
for (String receiver : notification.getReceivers()) {
SmsRequest request = new SmsRequest();
request.setPhone(receiver);
request.setTemplateId(templateId);
request.setParams(buildSmsParams(notification));
smsService.sendSms(request);
}
notification.setStatus(NotificationStatus.SENT);
notification.setSendTime(LocalDateTime.now());
logger.info("短信通知发送成功: {}", notification.getId());
} catch (Exception e) {
logger.error("短信通知发送失败: {}", notification.getId(), e);
notification.setStatus(NotificationStatus.FAILED);
throw new NotificationException("短信发送失败", e);
}
}
private Map<String, String> buildSmsParams(Notification notification) {
Map<String, String> params = new HashMap<>();
params.put("title", notification.getTitle());
params.put("content", notification.getContent());
return params;
}
@Override
public boolean supports(NotificationType type) {
return type == NotificationType.SMS;
}
}
// 站内信通知
public class SystemNotifier implements Notifier {
private static final Logger logger = LoggerFactory.getLogger(SystemNotifier.class);
private UserMessageRepository messageRepository;
public SystemNotifier(UserMessageRepository messageRepository) {
this.messageRepository = messageRepository;
}
@Override
public void send(Notification notification) {
try {
for (String receiver : notification.getReceivers()) {
UserMessage message = new UserMessage();
message.setUserId(receiver);
message.setTitle(notification.getTitle());
message.setContent(notification.getContent());
message.setRead(false);
message.setCreateTime(LocalDateTime.now());
messageRepository.save(message);
}
notification.setStatus(NotificationStatus.SENT);
notification.setSendTime(LocalDateTime.now());
logger.info("站内信通知发送成功: {}", notification.getId());
} catch (Exception e) {
logger.error("站内信通知发送失败: {}", notification.getId(), e);
notification.setStatus(NotificationStatus.FAILED);
throw new NotificationException("站内信发送失败", e);
}
}
@Override
public boolean supports(NotificationType type) {
return type == NotificationType.SYSTEM;
}
}
// 推送通知
public class PushNotifier implements Notifier {
private static final Logger logger = LoggerFactory.getLogger(PushNotifier.class);
private PushService pushService;
public PushNotifier(PushService pushService) {
this.pushService = pushService;
}
@Override
public void send(Notification notification) {
try {
for (String receiver : notification.getReceivers()) {
PushRequest request = new PushRequest();
request.setUserId(receiver);
request.setTitle(notification.getTitle());
request.setBody(notification.getContent());
pushService.push(request);
}
notification.setStatus(NotificationStatus.SENT);
notification.setSendTime(LocalDateTime.now());
logger.info("推送通知发送成功: {}", notification.getId());
} catch (Exception e) {
logger.error("推送通知发送失败: {}", notification.getId(), e);
notification.setStatus(NotificationStatus.FAILED);
throw new NotificationException("推送发送失败", e);
}
}
@Override
public boolean supports(NotificationType type) {
return type == NotificationType.PUSH;
}
}
通知管理器
@Service
public class NotificationManager {
private static final Logger logger = LoggerFactory.getLogger(NotificationManager.class);
private List<Notifier> notifiers;
private NotificationStrategy strategy;
private NotificationRepository repository;
private ExecutorService executorService;
private RetryPolicy retryPolicy;
@Autowired
public NotificationManager(List<Notifier> notifiers,
NotificationStrategy strategy,
NotificationRepository repository) {
this.notifiers = notifiers;
this.strategy = strategy;
this.repository = repository;
this.executorService = Executors.newFixedThreadPool(10);
this.retryPolicy = new SimpleRetryPolicy(3, 1000);
}
// 发送通知（异步）
public CompletableFuture<Notification> sendAsync(Notification notification) {
return CompletableFuture.supplyAsync(() -> {
try {
send(notification);
return notification;
} catch (Exception e) {
logger.error("异步通知发送失败: {}", notification.getId(), e);
throw new CompletionException(e);
}
}, executorService);
}
// 发送通知（同步）
public Notification send(Notification notification) {
List<Notifier> matchedNotifiers = strategy.select(notifiers, notification.getType());
for (Notifier notifier : matchedNotifiers) {
try {
notifier.send(notification);
repository.save(notification);
break;
} catch (Exception e) {
logger.error("通知发送异常，尝试下一个通知器", e);
notification.setRetryCount(notification.getRetryCount() + 1);
if (shouldRetry(notification)) {
retry(notification, notifier);
}
}
}
return notification;
}
// 批量发送通知
public List<Notification> sendBatch(List<Notification> notifications) {
List<CompletableFuture<Void>> futures = notifications.stream()
.map(n -> CompletableFuture.runAsync(() -> send(n), executorService))
.collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
return notifications;
}
// 重试机制
private void retry(Notification notification, Notifier notifier) {
logger.info("重试发送通知: {}，第{}次尝试", notification.getId(), notification.getRetryCount());
int attempt = 1;
while (attempt <= retryPolicy.getMaxRetries()) {
try {
Thread.sleep(retryPolicy.getBackoffMillis() * attempt);
notifier.send(notification);
notification.setStatus(NotificationStatus.SENT);
repository.save(notification);
return;
} catch (Exception e) {
logger.error("第{}次重试失败", attempt, e);
attempt++;
}
}
notification.setStatus(NotificationStatus.FAILED);
repository.save(notification);
}
private boolean shouldRetry(Notification notification) {
return notification.getRetryCount() < retryPolicy.getMaxRetries();
// 关闭资源
public void shutdown() {
executorService.shutdown();
}
}
策略模式实现
// 通知策略接口
public interface NotificationStrategy {
List<Notifier> select(List<Notifier> notifiers, NotificationType type);
}
// 单渠道策略
public class SingleChannelStrategy implements NotificationStrategy {
@Override
public List<Notifier> select(List<Notifier> notifiers, NotificationType type) {
return notifiers.stream()
.filter(n -> n.supports(type))
.limit(1)
.collect(Collectors.toList());
}
}
// 多渠道策略
public class MultiChannelStrategy implements NotificationStrategy {
@Override
public List<Notifier> select(List<Notifier> notifiers, NotificationType type) {
return notifiers.stream()
.filter(n -> n.supports(type))
.collect(Collectors.toList());
}
}
// 备用策略
public class FallbackStrategy implements NotificationStrategy {
@Override
public List<Notifier> select(List<Notifier> notifiers, NotificationType type) {
List<Notifier> matched = notifiers.stream()
.filter(n -> n.supports(type))
.collect(Collectors.toList());
if (matched.isEmpty()) {
// 备选所有通知器
return notifiers;
}
return matched;
}
}
重试策略
public class RetryPolicy {
private int maxRetries;
private long backoffMillis;
public RetryPolicy(int maxRetries, long backoffMillis) {
this.maxRetries = maxRetries;
this.backoffMillis = backoffMillis;
}
public int getMaxRetries() { return maxRetries; }
public long getBackoffMillis() { return backoffMillis; }
}
public class SimpleRetryPolicy extends RetryPolicy {
public SimpleRetryPolicy(int maxRetries, long backoffMillis) {
super(maxRetries, backoffMillis);
}
}
通知服务接口
public interface NotificationService {
void sendEmail(String to, String subject, String content);
void sendSms(String phone, String content);
void sendPush(String userId, String title, String content);
void sendSystemMessage(String userId, String title, String content);
}
@Service
public class NotificationServiceImpl implements NotificationService {
private NotificationManager notificationManager;
private ObjectMapper objectMapper;
@Autowired
public NotificationServiceImpl(NotificationManager notificationManager,
ObjectMapper objectMapper) {
this.notificationManager = notificationManager;
this.objectMapper = objectMapper;
}
@Override
public void sendEmail(String to, String subject, String content) {
Notification email = new Notification.Builder()
.id(UUID.randomUUID().toString())
.title(subject)
.content(content)
.receivers(Collections.singletonList(to))
.type(NotificationType.EMAIL)
.build();
notificationManager.sendAsync(email);
}
@Override
public void sendSms(String phone, String content) {
Notification sms = new Notification.Builder()
.id(UUID.randomUUID().toString())
.title("")
.content(content)
.receivers(Collections.singletonList(phone))
.type(NotificationType.SMS)
.build();
notificationManager.sendAsync(sms);
}
@Override
public void sendPush(String userId, String title, String content) {
Notification push = new Notification.Builder()
.id(UUID.randomUUID().toString())
.title(title)
.content(content)
.receivers(Collections.singletonList(userId))
.type(NotificationType.PUSH)
.build();
notificationManager.sendAsync(push);
}
@Override
public void sendSystemMessage(String userId, String title, String content) {
Notification systemMsg = new Notification.Builder()
.id(UUID.randomUUID().toString())
.title(title)
.content(content)
.receivers(Collections.singletonList(userId))
.type(NotificationType.SYSTEM)
.build();
notificationManager.sendAsync(systemMsg);
}
}
配置类
@Configuration
@EnableAsync
public class NotificationConfig {
@Bean
public NotificationStrategy notificationStrategy() {
return new MultiChannelStrategy(); // 或者使用其他策略
}
@Bean
public NotificationManager notificationManager(
List<Notifier> notifiers,
NotificationStrategy strategy,
NotificationRepository repository) {
return new NotificationManager(notifiers, strategy, repository);
}
@Bean
public EmailNotifier emailNotifier(JavaMailSender mailSender,
TemplateEngine templateEngine) {
return new EmailNotifier(mailSender, templateEngine);
}
@Bean
public SmsNotifier smsNotifier(SmsService smsService) {
return new SmsNotifier(smsService, "SMS_TEMPLATE_ID");
}
@Bean
public SystemNotifier systemNotifier(UserMessageRepository repository) {
return new SystemNotifier(repository);
}
@Bean
public PushNotifier pushNotifier(PushService pushService) {
return new PushNotifier(pushService);
}
@Bean
public ObjectMapper objectMapper() {
return new ObjectMapper()
.registerModule(new JavaTimeModule())
.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
}
}
异步处理组件
@Component
public class AsyncNotificationProcessor {
private static final Logger logger = LoggerFactory.getLogger(AsyncNotificationProcessor.class);
@Autowired
private NotificationManager notificationManager;
@Async("notificationExecutor")
public CompletableFuture<Notification> processAsync(Notification notification) {
return notificationManager.sendAsync(notification);
}
@Bean(name = "notificationExecutor")
public Executor notificationExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(20);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("notification-");
executor.initialize();
return executor;
}
}
监控和记录
@Component
public class NotificationMonitor {
private static final Logger logger = LoggerFactory.getLogger(NotificationMonitor.class);
private Map<String, Long> notificationCount = new ConcurrentHashMap<>();
private Map<String, Long> notificationFailures = new ConcurrentHashMap<>();
public void recordSuccess(Notification notification) {
String type = notification.getType().name();
notificationCount.merge(type, 1L, Long::sum);
logger.info("Notification success: {}, type: {}, cost: {}ms",
notification.getId(),
notification.getType(),
System.currentTimeMillis() - notification.getCreateTime().getNano() / 1_000_000);
}
public void recordFailure(Notification notification) {
String type = notification.getType().name();
notificationFailures.merge(type, 1L, Long::sum);
logger.warn("Notification failure: {}, type: {}",
notification.getId(),
notification.getType());
}
public Map<String, Long> getNotificationStats() {
Map<String, Long> stats = new HashMap<>();
stats.putAll(notificationCount);
stats.put("failures", notificationFailures.values().stream().mapToLong(Long::longValue).sum());
return stats;
}
}
使用示例
@RestController
@RequestMapping("/api/notifications")
public class NotificationController {
@Autowired
private NotificationService notificationService;
@Autowired
private AsyncNotificationProcessor asyncProcessor;
// 发送邮件通知
@PostMapping("/email")
public ResponseEntity<?> sendEmail(@RequestBody EmailRequest request) {
notificationService.sendEmail(
request.getTo(),
request.getSubject(),
request.getContent()
);
return ResponseEntity.ok().body("邮件通知发送成功");
}
// 发送短信通知
@PostMapping("/sms")
public ResponseEntity<?> sendSms(@RequestBody SmsRequest request) {
notificationService.sendSms(request.getPhone(), request.getContent());
return ResponseEntity.ok().body("短信通知发送成功");
}
// 发送推送通知
@PostMapping("/push")
public ResponseEntity<?> sendPush(@RequestBody PushRequest request) {
notificationService.sendPush(
request.getUserId(),
request.getTitle(),
request.getContent()
);
return ResponseEntity.ok().body("推送通知发送成功");
}
// 批量发送
@PostMapping("/batch")
public ResponseEntity<?> sendBatch(@RequestBody BatchNotificationRequest request) {
List<Notification> notifications = request.getNotifications();
notificationManager.sendBatch(notifications);
return ResponseEntity.ok().body("批量通知发送完成");
}
// 获取通知统计
@GetMapping("/stats")
public ResponseEntity<Map<String, Long>> getStats() {
return ResponseEntity.ok(notificationMonitor.getNotificationStats());
}
}
// 请求DTO
public class EmailRequest {
private String to;
private String subject;
private String content;
// getters and setters...
}
public class SmsRequest {
private String phone;
private String content;
// getters and setters...
}
public class PushRequest {
private String userId;
private String title;
private String content;
// getters and setters...
}
public class BatchNotificationRequest {
private List<Notification> notifications;
// getters and setters...
}
异常处理
@RestControllerAdvice
public class GlobalExceptionHandler {
@ExceptionHandler(NotificationException.class)
public ResponseEntity<ErrorResponse> handleNotificationException(NotificationException e) {
return ResponseEntity
.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body(new ErrorResponse(e.getMessage(), System.currentTimeMillis()));
}
@ExceptionHandler(ValidationException.class)
public ResponseEntity<ErrorResponse> handleValidationException(ValidationException e) {
return ResponseEntity
.status(HttpStatus.BAD_REQUEST)
.body(new ErrorResponse(e.getMessage(), System.currentTimeMillis()));
}
}
public class NotificationException extends RuntimeException {
public NotificationException(String message) {
super(message);
}
public NotificationException(String message, Throwable cause) {
super(message, cause);
}
}
public class ErrorResponse {
private String message;
private long timestamp;
public ErrorResponse(String message, long timestamp) {
this.message = message;
this.timestamp = timestamp;
}
// getters and setters...
}
主要特性：
- 策略模式：灵活选择通知渠道
- 建造者模式：简化对象创建
- 异步处理：使用CompletableFuture和线程池
- 重试机制：自动重试失败的通知
- 监控统计：记录通知发送情况
- 多渠道支持：邮件、短信、推送、站内信
- 可扩展性：易于添加新的通知方式
这个案例展示了如何构建一个健壮、可扩展的通知系统，适合各种实际应用场景。
