背景
公司有个主系统 A，这个系统将来用途作为 SaaS 服务系统，设计尽可能保持纯粹，不要引入过多的中间件技术
其下有 B,C 甚至未来还有多个子系统，现在的需求是需要当子系统 B,C 开启了某个配置，需要对接 A 系统，A 系统能够提供 B,C 系统的能力赋能，当 BC 断开配置，B,C 又是个纯粹可以单独运行的系统，某个系统不共用同一套数据，拥有自己的环境和数据，当子系统开启了配置，相关接口功能需要限制，由主系统提供能力，断开配置，A.B,C 三个系统能够独立运行
功能拆分
主系统:
对主系统相关的增量和全量接口数据进行同步传输：这里有两个关键点
- 有许多增量接口需要进行同步传输&&传输数据的逻辑不能影响原先的业务逻辑（通过 AOP 注解）
- 需要找到一个类型消息队列 mq 那种形势进行数据同步，A 系统数据发生改动子系统能够感知
但是这个系统不能引入过多的中间件，最终经过调研，选择使用了 Spring 的 STOMP 和 Spring event 事件实现一种伪消息队列
子系统：
- 子系统相关接口限制：通过 AOP 接口限制
- 子系统需要接收主系统的全量和增量数据 （STOMP+AOP+SDK）
最终将功能细节拆分后，对于这个需求设计选型最终采用：
Spring STOMP+ Spring EVENT+AOP+SDK+ 线程池进行业务线程隔离
时序图
- 全量数据
- 增量数据同步
设计伪消息队列
客户端
TenantClient
用来与 A 系统进行对接的客户端
send 方法
通过这方法客户端也就是目前的 B,C 系统可以主动向 A 系统发送消息
- path: 发送的接口路径
- payload: 发送的消息体
- stompSession：Spring stomp 的与服务端的连接对象
public void send(String path, Object payload) throws DisconnectException {
if (stompSession == null) {
throw new DisconnectException();
} else {
stompSession.send(APPLICATION_DESTINATION_PREFIX + path, payload);
}
}
subscribe 方法
进行客户端订阅事件，与服务端进行绑定
/**
* 订阅指定路径的事件，可以多次订阅
*
* @param path 订阅的路径
* @param observer 消费者
* @param <T> 消息类型
*/
@SuppressWarnings("unchecked")
public <T> void subscribe(String path, Observer<T> observer) {
subscribe(path, new MessageHandler() {
@Override
public Type getType() {
return observer.getType();
}
@Override
public void handle(Object payload) {
observer.handle((T) payload);
}
});
}
/**
* 订阅指定路径的事件，可以多次订阅
*
* @param path 订阅的路径
* @param handler 消费者
*/
public void subscribe(String path, MessageHandler handler) {
subscriptions.compute(path, (key, oldValue) -> {
List<TenantSubscription> value = oldValue == null ? new CopyOnWriteArrayList<>() : oldValue;
value.add(new TenantSubscription(handler));
return value;
});
}
private final ConcurrentMap<String, List<TenantSubscription>> subscriptions = new ConcurrentHashMap<>();
通过一个 ConcurrentMap 容器存储 key:客户端接口路径 value:消费者列表 支持一个路径对应多个客户端，防止读写并发 使用 CopyOnWriteArrayList
start 方法:
容器启动使用使用 Spring 事件 调用 start 方法
@EventListener
public void onAppStarted(ApplicationStartedEvent ignored) {
tenantClient().start();
}
destory 方法:
容器销毁执行
@PreDestroy
public void destroy() {
tenantClient().stop();
}
public class TenantClient {
private static final Logger logger = LoggerFactory.getLogger(TenantClient.class);
public static final String APPLICATION_DESTINATION_PREFIX = "/app";
public static final String TOPIC_PREFIX = "/topic";
public static final String DATE_TIME_FORMATTER_PATTERN = "yyyy-MM-dd HH🇲🇲ss";
/** 配置信息 */
private final TenantConfig config;
/** STOMP客户端 */
private final StompClient client;
/** 订阅映射: 订阅路径 -> 消费者列表 */
private final ConcurrentMap<String, List<TenantSubscription>> subscriptions = new ConcurrentHashMap<>();
private Runnable onConnectedListener;
/** 是否已开启 */
private final AtomicBoolean started = new AtomicBoolean();
/** 当前与服务端的连接 */
private StompSession stompSession;
/** LDAP数据源 */
private BaseLdapPathContextSource contextSource;
private ObjectMapper objectMapper = createObjectMapper();
public TenantClient(TenantConfig config) {
this.config = Objects.requireNonNull(config);
String url = config.getBaseUrl() + config.getPath();
this.client = new StompClient(url, new TenantStompSessionHandler(), config)
.setReconnectInterval(config.getReconnectInterval());
setObjectMapper(objectMapper);
}
/**
* 配置用于传输数据的JSON序列化对象
*
* @param objectMapper objectMapper
*/
public void setObjectMapper(ObjectMapper objectMapper) {
this.objectMapper = Objects.requireNonNull(objectMapper);
MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
converter.setObjectMapper(objectMapper);
this.client.setMessageConverter(converter);
}
/**
* 配置LDAP数据源
*
* @param contextSource LDAP数据源
*/
public void setContextSource(BaseLdapPathContextSource contextSource) {
this.contextSource = contextSource;
}
/**
* 配置任务调度器，用于心跳检测
*
* @param taskScheduler 任务调度器
*/
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.client.setTaskScheduler(taskScheduler);
}
/**
* 开始连接
*
* @return 连接结果Future
*/
@SuppressWarnings({"unchecked", "rawtypes"})
public ListenableFuture<Object> start() {
if (!started.compareAndSet(false, true)) {
throw new IllegalStateException("不能重复启动");
}
return (ListenableFuture) client.connect();
}
/**
* 关闭客户端，断开连接
*/
public void stop() {
if (started.compareAndSet(true, false)) {
if (stompSession != null) {
stompSession.disconnect();
}
stompSession = null;
client.stop();
}
}
/**
* 订阅指定路径的事件，可以多次订阅
*
* @param path 订阅的路径
* @param observer 消费者
* @param <T> 消息类型
*/
@SuppressWarnings("unchecked")
public <T> void subscribe(String path, Observer<T> observer) {
subscribe(path, new MessageHandler() {
@Override
public Type getType() {
return observer.getType();
}
@Override
public void handle(Object payload) {
observer.handle((T) payload);
}
});
}
/**
* 订阅指定路径的事件，可以多次订阅
*
* @param path 订阅的路径
* @param handler 消费者
*/
public void subscribe(String path, MessageHandler handler) {
subscriptions.compute(path, (key, oldValue) -> {
List<TenantSubscription> value = oldValue == null ? new CopyOnWriteArrayList<>() : oldValue;
value.add(new TenantSubscription(handler));
return value;
});
}
/**
* 取消指定路径的订阅
*
* @param path 取消订阅的路径
* @return 是否有消费者被取消
*/
public boolean unsubscribe(String path) {
List<TenantSubscription> list = subscriptions.remove(path);
if (list == null) return false;
for (TenantSubscription it : list) {
if (it.getSubscription() != null) {
it.getSubscription().unsubscribe();
}
}
return true;
}
/**
* 发送消息到租户融合中心
*
* @param path 消息路径
* @param payload 消息内容
* @throws DisconnectException 如果未连接到客户端
*/
public void send(String path, Object payload) throws DisconnectException {
if (stompSession == null) {
throw new DisconnectException();
} else {
stompSession.send(APPLICATION_DESTINATION_PREFIX + path, payload);
}
}
/**
* 用户身份认证
*
* @param username 用户名
* @param password 密码
* @return true为认证通过，false为用户名或密码不正确
*/
public boolean auth(String username, String password) {
try {
Name dn = contextSource.getBaseLdapName()
.add(config.getUserBase())
.add(MessageFormat.format(config.getUserRdn(), username));
contextSource.getContext(dn.toString(), password);
return true;
} catch (NamingException ex) {
if ((ex instanceof org.springframework.ldap.AuthenticationException)
|| (ex instanceof org.springframework.ldap.OperationNotSupportedException)) {
return false;
} else {
throw ex;
}
} catch (InvalidNameException e) {
throw new IllegalArgumentException(e);
}
}
public void onConnected(Runnable onConnectedListener) {
this.onConnectedListener = onConnectedListener;
}
public boolean isConnected() {
return stompSession != null;
}
private static ObjectMapper createObjectMapper() {
ObjectMapper mapper = new ObjectMapper();
mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
mapper.disable(SerializationFeature.WRITE_DATE_TIMESTAMPS_AS_NANOSECONDS);
// 设置默认日期格式
mapper.setDateFormat(new SimpleDateFormat(DATE_TIME_FORMATTER_PATTERN));
mapper.registerModule(new JavaTimeModule());
// 修改LocalDateTime的默认格式
mapper.registerModule(createTimeModule());
return mapper;
}
private static SimpleModule createTimeModule() {
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(DATE_TIME_FORMATTER_PATTERN);
SimpleModule timeModule = new SimpleModule();
timeModule.addDeserializer(LocalDateTime.class, new LocalDateTimeDeserializer(dateTimeFormatter));
timeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(dateTimeFormatter));
return timeModule;
}
private class TenantStompSessionHandler extends StompSessionHandlerAdapter {
@Override
public void afterConnected(@NonNull StompSession session, @NonNull StompHeaders connectedHeaders) {
logger.info("与租户融合中心建立连接: {}", config.getBaseUrl());
stompSession = session;
subscriptions.forEach((path, list) -> {
for (TenantSubscription subscription : list) {
registerSubscription(session, path, subscription);
}
});
if (onConnectedListener != null) {
try {
onConnectedListener.run();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
private void registerSubscription(StompSession session, String path, TenantSubscription subscription) {
StompSession.Subscription sub = session.subscribe(TOPIC_PREFIX + path, new StompFrameHandler() {
@Override
public @NonNull Type getPayloadType(@NonNull StompHeaders headers) {
Type type = subscription.getHandler().getType();
return type instanceof ParameterizedType ? byte[].class : type;
}
@Override
public void handleFrame(@NonNull StompHeaders headers, Object payload) {
String path = getPath(headers);
String errorMessage = null;
try {
Type type = subscription.getHandler().getType();
if (type instanceof ParameterizedType) {
// 处理泛型参数
payload = objectMapper.readValue((
byte[]) payload, TypeFactory.defaultInstance().constructType(type));
}
subscription.getHandler().handle(payload);
} catch (Exception e) {
logger.error("处理消息[{}]出错. payload: {}, error: {}",
path, payload, e.getMessage(), e);
errorMessage = e.getMessage();
} finally {
sendClientLog(path, payload, errorMessage);
}
}
});
subscription.setSubscription(sub);
}
@Override
public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, Throwable exception) {
logger.error("handleException: {}", exception.getMessage(), exception);
}
@Override
public void handleTransportError(StompSession session, Throwable exception) {
if (exception instanceof ConnectionLostException) {
stompSession = null;
}
logger.error("handleTransportError: {}", exception.getMessage(), exception);
}
private String getPath(StompHeaders headers) {
String destination = Objects.requireNonNull(headers.getDestination());
return destination.startsWith(TOPIC_PREFIX) ? destination.substring(TOPIC_PREFIX.length()) : destination;
}
private void sendClientLog(String path, Object payload, String errorMessage) {
try {
LogEvent logEvent = new LogEvent();
if (errorMessage == null) {
logEvent.setContent(String.format("处理消息[%s]成功", path));
} else {
logEvent.setContent(String.format("处理消息[%s]失败: %s. payload: %s", path, errorMessage, payload));
}
logEvent.setSuccess(errorMessage == null);
logEvent.setCreateTime(LocalDateTime.now());
send(TenantPaths.CLIENT_LOG, logEvent);
} catch (Exception e) {
logger.error("发送调用日志出错: {}", e.getMessage(), e);
}
}
}
@RequiredArgsConstructor
@Getter
@Setter
private static class TenantSubscription {
private final MessageHandler handler;
private StompSession.Subscription subscription;
}
}
使用 @ConditionalOnProperty 注解，只有当配置文件开关开启，才会加载这个配置类，才会初始化 TenantClient 这个 Bean
@ConditionalOnProperty(value = "xxxx.enable")
@ TenantObserver 注解
客户端方法添加该接口，可以接收到生产者订阅增量接口发送的数据
@TenantObserver(PULL_PROJECT_MEMBER_REQUEST)
public void pullSyncProjectMemberData(List<ProjectMemberDto> syncProjectMemberDtoList) {
Map<Integer, List<UserProjectPermission>> existingProjectMembersMap = projectPermissionService.findAll()
.stream()
.collect(Collectors.groupingBy(UserProjectPermission::getUserId));
Map<String, List<ProjectMemberDto>> incomingUsersMap = syncProjectMemberDtoList
.stream()
.collect(Collectors.groupingBy(ProjectMemberDto::getUsername));
Map<String, Integer> idAndNameUserMap = userService2.findByUsernames(incomingUsersMap.keySet())
.stream()
.collect(Collectors.toMap(User::getUserName, User::getId));
Map<Integer, List<ProjectMemberDto>> idIncomingUsersMap = incomingUsersMap.entrySet()
.stream()
.filter(entry -> idAndNameUserMap.containsKey(entry.getKey()))
.collect(Collectors.toMap(
entry -> idAndNameUserMap.get(entry.getKey()),
Map.Entry::getValue,
(existing, replacement) -> replacement));
CompletableFuture.runAsync(() -> processNewAndUpdatedPermissions(idIncomingUsersMap));
processDeletePermission(existingProjectMembersMap, idIncomingUsersMap.keySet());
}
消费者接口幂等性
因为生产者可能迟迟等不到消费者的确认 confirm 反馈，会不断重试轮询消息表，消费者接口需要保证数据的一致性，会将当前消费的消息存储起来，数据库存在该消息，则直接返回 confirm 确认
消费者成功消费消息后，callBack 回调 tenant.send 方法给生产者，返回当前事务 id 和 confirm 确认机制
消费者如何绑定 subscribe 方法呢？
客户端订阅接口需要绑定发送，利用 AOP 注解或者实现 BeanPostProcessor 这个 Spring bean 后置处理器
@RequiredArgsConstructor
public class TenantObserverBeanProcessor implements BeanPostProcessor {
private final TenantClient client;
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
if (AnnotationUtils.isCandidateClass(targetClass, TenantObserver.class)) {
MethodIntrospector.selectMethods(
targetClass,
(MethodIntrospector.MetadataLookup<TenantObserver>) method ->
AnnotatedElementUtils.findMergedAnnotation(method, TenantObserver.class))
.forEach((method, tenantObserver) -> subscribeObserver(bean, method, tenantObserver));
}
return bean;
}
private void subscribeObserver(Object bean, Method method, TenantObserver tenantObserver) {
Type[] parameterTypes = method.getGenericParameterTypes();
if (parameterTypes.length > 1) {
throw new IllegalStateException(method + "方法只能有一个参数");
}
Type payloadType = parameterTypes[0];
client.subscribe(tenantObserver.value(), new MessageHandler() {
@Override
public Type getType() {
return payloadType;
}
@Override
public void handle(Object payload) throws Exception {
try {
if (method.getParameterCount() == 0) {
method.invoke(bean);
} else {
method.invoke(bean, payload);
}
} catch (InvocationTargetException e) {
Throwable target = e.getTargetException();
if (target instanceof Exception) {
throw (Exception) target;
}
throw e;
}
}
});
}
}
会再 bean 初始化前，执行扫描 @TenantObserver 注解，替换成代理类，进行方法增强，实现路径的绑定
生产者
发送消息 messagingTemplate
生产者需要生产的消息的时候，通过 messagingTemplate 进行 convertAndSend 进行消息的发送到客户端绑定的路径
public void send(String path, Object payload) {
messagingTemplate.convertAndSend(WebSocketConstants.TOPIC_PREFIX + path, payload);
}
WebSocketConstants.xx 路径需要和客户端 subscribe 路径绑定 key 路径保持一致
业务隔离
public void syncxxxx(UserDto dto, Optional<UserDto> oldDtoOp, String creator) {
if (Objects.equals(dto.getUsername(), oldDtoOp.map(UserDto::getUsername).orElse(null))) {
return;
}
List<Long> projectCodes = projectRepository.findCodeListByTenantName(dto.getTenant());
if (CollUtil.isEmpty(projectCodes)) {
return;
}
oldDtoOp.map(UserDto::getUsername).ifPresent(projectRepository::deleteByUserName);
List<ProjectMemberDto> inserList = projectCodes
.stream()
.map(code -> ProjectMemberDto.valueOf(code, dto.getUsername(), ProjectPermissionType.ADMIN.getCode(), creator))
.collect(Collectors.toList());
saveBatchOrUpdateProjectMember(inserList);
//发送send事件
eventPublisher.publishEvent(new ProjectMemberEvent(this, inserList,
oldDtoOp.map(UserDto::getUsername).map(Set::of).orElseGet(Collections::emptySet), projectCodes));
}
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
@Async(SYNC_DATA_EXECUTOR)
public void send(ProjectMemberEvent event) {
//进行发送数据
messagingTemplate.convertAndSend(xxxxxx)
}
使用线程池天然的线程隔离技术和 Spring 事件 进行业务数据和发送消息解耦，实现 mq 消息队列的业务解耦功能
因为使用了多线程，所以 Spring 事务注解无法生效，这里使用了 @TransactionalEventListener 注解，当事务提交后才会触发
生产者消息持久性
在生产者发送数据之前，会将当前的数据产生快照信息，进行数据库入库，增加一个是否被消费的标识，客户端消费后返回一个 confirm 和当前事务 id 进行确认机制，将当前消息删除 （短暂的数据不一致性，后续有全量同步补偿机制）
消费者向生产者全量拉取数据
- 程序容器启动会主动拉取一次
- 定时调度会拉取一次
@EventListener
public void onTenantConnected(TenantConnectedEvent ignore) {
if (isFirst.compareAndSet(true, false)) {
try {
executorService.submit(() -> beginSendPullRequest());
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
private void beginSendPullRequest() throws DisconnectException {
tenantClient.send(PULL_REQUEST, "");
tenantClient.send(PULL_PROJECT_REQUEST, "");
tenantClient.send(PULL_QUEUE_REQUEST, "");
tenantClient.send(PULL_PROJECT_MEMBER_REQUEST, "");
}
@Scheduled(cron = "${sync-data-cron}")
public void schedulePullData() {
if (enable && tenantClient.isConnected()) {
try {
beginSendPullRequest();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
补偿机制
- 增量定时拉取数据
- 日志监控
使用 AOP 注解收集日志上报
@Aspect
@Component
@RequiredArgsConstructor
public class MessageMappingAspect {
private static final Logger logger = LoggerFactory.getLogger(MessageMappingAspect.class);
private final DockedClientDomainService dockedClientDomainService;
@Around("@annotation(org.springframework.messaging.handler.annotation.MessageMapping)")
public Object handle(ProceedingJoinPoint point) throws Throwable {
String errorMessage = null;
try {
return point.proceed();
} catch (Exception e) {
errorMessage = e.getMessage();
throw e;
} finally {
WebSocketMessageHolder message = WebSocketMessageHolder.get().orElse(null);
try {
if (message != null && message.getDestination() != null &&
!message.getDestination().endsWith(TenantPaths.CLIENT_LOG)) {
addLog(message, errorMessage);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
private void addLog(WebSocketMessageHolder message, String errorMessage) {
var event = new LogEvent();
var path = CharSequenceUtil.removePrefix(message.getDestination(), WebSocketConstants.APPLICATION_DESTINATION_PREFIX);
String content;
if (errorMessage == null) {
content = String.format("发送消息[%s]成功", path);
} else {
content = String.format("发送消息[%s]失败: %s. payload: %s", path, errorMessage, getPayload(message));
}
event.setContent(content);
event.setSuccess(errorMessage == null);
event.setCreateTime(LocalDateTime.now());
dockedClientDomainService.addLog(message.getUser().getName(), event);
}
private static Object getPayload(WebSocketMessageHolder message) {
if (message.getPayload() instanceof byte[] bytes) {
return new String(bytes);
}
return message.getPayload();
}
}
- 客户端连接监控
结论
通过这个伪消息队列，可以解决我们现在的需求，但是没法跟真正的 mq 消息队列相比，也无法做到数据实时的一致性，只能是数据的最终一致性加上补偿机制（增量 + 全量），无法做到消息队列的系统层次的解耦和流量控制
**优点也有，不需要引入其他的中间，只需要 spring+ 线程池就可以实现，运维成本减少，有利有弊，**根据业务选型技术