做dw和ps的网站教学,布料市场做哪个网站好,建网站方案书,舟山建设银行网站文章目录 依赖bus应用接口用到的封装参数类 接收的应用监听器定义的事件类 使用bus定义bus远程调用A应用数据更新后通过bus数据同步给B应用 依赖 dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-bus-amqp… 文章目录 依赖bus应用接口用到的封装参数类 接收的应用监听器定义的事件类 使用bus定义bus远程调用A应用数据更新后通过bus数据同步给B应用 依赖 dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-bus-amqp/artifactId/dependencybus应用
类似于生产者
接口
供内部其他应用使用,远程调用该接口实现各应用之间数据同步参数1定义事件,参数2定义操作具体crud,参数3定义传参数据,参数4定义给哪个应用(nacos注册的应用名)同步数据
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.xyc.sms.common.bus.events.DataSyncEventEnum;
import com.xyc.sms.common.bus.events.DataSyncEventFactory;
import com.xyc.sms.common.bus.events.DataSyncOperateTypeEnum;
import com.xyc.sms.common.entity.Result;
import org.springframework.cloud.bus.ServiceMatcher;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;/*** 数据同步通知事件控制器,该事件主要用于平台中常规的数据同步通知,* 需要用于其他功能请新增类* 需要增加同步的方式请在{link DataSyncEventEnum}增加事件枚举* 同时在{link com.xyc.sms.common.bus.events.dataSync}下增加事件类,新增的事件类需要继承{link com.xyc.sms.common.bus.events.DataSyncEvent}*/
RestController
RequestMapping(/default)
public class DataSyncNotifyEventController {private static final Log logger LogFactory.get();Resourceprivate ServiceMatcher busServiceMatcher;Resourceprivate ApplicationEventPublisher applicationEventPublisher;/*** 发布数据同步通知事件** param eventEnum 事件枚举,可通过枚举找到对应的事件类* param operateType 操作类型枚举* param obj 需要处理的消息* param destination 目的地,为null则是广播给所有该事件的监听器* return 发布结果*/PostMapping(/publish/{eventEnum}/{operateType})public Result publishDataSyncNotifyEvent(PathVariable(eventEnum) DataSyncEventEnum eventEnum,PathVariable(operateType) DataSyncOperateTypeEnum operateType,RequestBody Object obj,RequestParam(value destination, required false) String destination) {try {applicationEventPublisher.publishEvent(DataSyncEventFactory.getInstanceForEvent(eventEnum,operateType,obj,busServiceMatcher.getServiceId(),destination));return Result.returnSuccessWithMsg(success);} catch (Exception e) {logger.error(e);return Result.returnFail(e.getMessage());}}
}事件工厂类
import com.fasterxml.jackson.databind.ObjectMapper;import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.Arrays;public class DataSyncEventFactory {private static final ObjectMapper OM new ObjectMapper();/*** 通过事件的类模板获取构造器并调用,生成事件的实体类** param operateType 操作类型* param source 事件原数据* param originService 原服务* param destinationService 目标服务* return 事件实体类* throws NoSuchMethodException 通过类模板无法找到相应的构造方法所抛出的异常* throws InvocationTargetException 构造器创建实例可能出现的调用目标异常* throws InstantiationException 构造器创建实例可能出现的实例化异常* throws IllegalAccessException 构造器创建实例可能出现的无法访问异常* throws IOException json转化出现IO的异常* throws ClassNotFoundException 通过类名{link DataSyncEventEnum#getEventClassName()}没有找到对应类*/public static DataSyncEvent? getInstanceForEvent(DataSyncEventEnum eventEnum,DataSyncOperateTypeEnum operateType,Object source,String originService,String destinationService)throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, IOException, ClassNotFoundException {Constructor?[] constructors DataSyncEventFactory.getEventClass(eventEnum).getDeclaredConstructors();Constructor? constructor Arrays.stream(constructors).filter(c - c.getParameterCount() 4).findFirst().orElseThrow(NoSuchMethodException::new);// 值转化Object o OM.readValue(OM.writeValueAsString(source), constructor.getParameterTypes()[1]);return (DataSyncEvent?) constructor.newInstance(operateType, o, originService, destinationService);}private static Class? getEventClass(DataSyncEventEnum eventEnum) throws ClassNotFoundException {return Class.forName(eventEnum.getEventClassName());}
}用到的封装参数类 定义各事件的枚举(只定义全类限定名称)
public enum DataSyncEventEnum {/*** 黑名单同步,参数为事件的类型名注意需要使用全限定类名* see com.xyc.sms.common.bus.events.dataSync.BlackListSyncEvent*/BLACKLIST_SYN(com.xyc.sms.common.bus.events.dataSync.BlackListSyncEvent),/*** 路由同步* see com.xyc.sms.common.bus.events.dataSync.RouteSyncEvent*/ROUTE_SYN(com.xyc.sms.common.bus.events.dataSync.RouteSyncEvent);/*** 事件类型名* see DataSyncEvent 该抽象类的实现类*/private final String eventClassName;DataSyncEventEnum(String eventClassName) {this.eventClassName eventClassName;}public String getEventClassName() {return eventClassName;}
}定义crud操作枚举
public enum DataSyncOperateTypeEnum implements Serializable {ADD, UPD, DEL
}推送的事件类
/*** 数据同步通知事件,作为一般通用事件使用,如需要特殊处理建议新增事件*/
public abstract class DataSyncEventT extends RemoteApplicationEvent {/*** 事件数据*/private DataSyncT dataSync;public DataSyncT getDataSync() {return dataSync;}public void setDataSync(DataSyncT dataSync) {this.dataSync dataSync;}/*** 基础构造器** param source 引发事件的原始数据* param originService 引发事件的原始服务* param destinationService 事件的目标服务*/public DataSyncEvent(DataSyncT source, String originService, String destinationService) {super(source, originService, destinationService);this.dataSync source;}/*** 事件的日志打印,会在监听器监听到事件时输出打印* 结果尽可能不要有换行,保证日志输出在一行内* 该方法可以在子类中重写** return 日志*/public String logPrint() {return String.format({\originService\:\%s\,\destinationService\:\%s\,\id\:\%s\,\dataSync\:%s,\timestamp\:\%s\}, this.getId(), this.getOriginService(), this.getDestinationService(), Objects.nonNull(this.dataSync) ? this.dataSync.toString() : null, this.getTimestamp());}/*** 数据同步的原始数据封装*/public static class DataSyncT implements Serializable {private DataSyncOperateTypeEnum operateType;private T data;public DataSync() {}public DataSync(DataSyncOperateTypeEnum operateType, T data) {this.operateType operateType;this.data data;}public DataSyncOperateTypeEnum getOperateType() {return operateType;}public T getData() {return data;}Overridepublic String toString() {return {\operateType\: operateType ,\data\: data };}}
}应用枚举
/*** 服务枚举*/
public enum ServiceEnum {SMS_BLACK_API(sms-black-api),SMS_RULES(sms-rules);public final String serviceName;ServiceEnum(String serviceName) {this.serviceName serviceName;}
}接收的应用
类似于消费者
监听器
推送过来的操作枚举类crud的值,决定执行哪个crud的具体方法该类放在接收的应用中,其他顶部继承的类放在common包中即可
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.xyc.sms.common.bus.DataSyncListener;
import com.xyc.sms.common.bus.events.dataSync.RouteSyncEvent;
import com.xyc.sms.common.entity.sms.Route;
import com.xyc.sms.rules.dao.boss.RouteMapper;
import com.xyc.sms.rules.data.RuleSymbol;
import com.xyc.sms.rules.service.SynService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;/*** 路由同步通知监听器*/
Component
public class RouteSynNotifyListener extends DataSyncListenerRouteSyncEvent, ListRoute {private static final Log log LogFactory.get();Autowiredprivate RouteMapper routeMapper;Autowiredprivate SynService synService;Overridepublic void handleByADD(ListRoute data) {Optional.ofNullable(data).ifPresent(ls - {if (ls.isEmpty()) {return;}long l System.currentTimeMillis();String time DateUtil.formatDateTime(ls.get(0).getCreateTime());ListRoute list routeMapper.selectByCreatetime(time);if (CollectionUtil.isEmpty(list)) {return;}// 加载到内存中list.forEach(r - RuleSymbol.RouteMap.put(r.getId(), r));synService.transformRoute(list, (s, r) - RuleSymbol.RouteChannelMap.put(s, r));log.info(RouteSynNotifyListener - {} - add | createTime:{}, (System.currentTimeMillis() - l), time);});}Overridepublic void handleByUPD(ListRoute data) {// 流程还是先删除后新增的方式Optional.ofNullable(data).ifPresent(ls - {ListInteger collect ls.stream().map(Route::getId).filter(Objects::nonNull).collect(Collectors.toList());if (collect.isEmpty()) {return;}long l System.currentTimeMillis();// 如果有先删除ListRoute c collect.stream().map(RuleSymbol.RouteMap::get).filter(Objects::nonNull).collect(Collectors.toList());if (!c.isEmpty()) {synService.transformRoute(c, (s, r) - RuleSymbol.RouteChannelMap.remove(s, r));}// 如果有再加入ListRoute list routeMapper.selectById(collect);if (!list.isEmpty()) {list.forEach(r - RuleSymbol.RouteMap.put(r.getId(), r));synService.transformRoute(list, (s, r) - RuleSymbol.RouteChannelMap.put(s, r));}log.info(RouteSynNotifyListener - {} - update | {}, (System.currentTimeMillis() - l), collect);});}Overridepublic void handleByDEL(ListRoute data) {Optional.ofNullable(data).ifPresent(ls - {long l System.currentTimeMillis();ListRoute collect ls.stream().map(r - RuleSymbol.RouteMap.remove(r.getId())).filter(Objects::nonNull).collect(Collectors.toList());if (collect.isEmpty()) {return;}synService.transformRoute(collect, (s, r) - RuleSymbol.RouteChannelMap.remove(s));log.info(RouteSynNotifyListener - {} - remove, (System.currentTimeMillis() - l));});}
}继承的抽象监听类
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.xyc.sms.common.bus.events.DataSyncEvent;
import org.springframework.context.ApplicationListener;/*** 数据同步事件监听器* 需要具体的子类实现,并注册到spring容器中** param T 数据同步件*/
public abstract class DataSyncListenerT extends DataSyncEventD, D implements ApplicationListenerT {private static final Log logger LogFactory.get();Overridepublic void onApplicationEvent(T event) {logger.info([DataSyncListener][onApplicationEvent] trigger event - {} - {}, event.getClass().getName(), event.logPrint());try {triggerEvent(event);} catch (Exception e) {logger.error(e);}}/*** 触发监听,处理事件** param event 事件*/public void triggerEvent(T event) {DataSyncEvent.DataSyncD source event.getDataSync();switch (source.getOperateType()) {case ADD:handleByADD(source.getData());return;case UPD:handleByUPD(source.getData());return;case DEL:handleByDEL(source.getData());return;default:}}/*** 处理添加事件* 由子类实现** param data 需要处理的数据*/public abstract void handleByADD(D data);/*** 处理修改事件* 由子类实现** param data 需要处理的数据*/public abstract void handleByUPD(D data);/*** 处理删除事件* 由子类实现** param data 需要处理的数据*/public abstract void handleByDEL(D data);
}定义的事件类
上面封装的事件枚举所记录的是该类的全类限定名称
package com.xyc.sms.common.bus.events.dataSync;import com.xyc.sms.common.bus.events.DataSyncEvent;
import com.xyc.sms.common.bus.events.DataSyncOperateTypeEnum;
import com.xyc.sms.common.entity.sms.Route;import java.util.List;public class RouteSyncEvent extends DataSyncEventListRoute {private static final long serialVersionUID -501657066268464154L;public RouteSyncEvent(DataSyncOperateTypeEnum operateType, ListRoute Routes, String originService, String destinationService) {super(new DataSync(operateType, Routes), originService, destinationService);}
}继承的抽象事件类
/*** 数据同步通知事件,作为一般通用事件使用,如需要特殊处理建议新增事件*/
public abstract class DataSyncEventT extends RemoteApplicationEvent {/*** 事件数据*/private DataSyncT dataSync;public DataSyncT getDataSync() {return dataSync;}public void setDataSync(DataSyncT dataSync) {this.dataSync dataSync;}/*** 基础构造器** param source 引发事件的原始数据* param originService 引发事件的原始服务* param destinationService 事件的目标服务*/public DataSyncEvent(DataSyncT source, String originService, String destinationService) {super(source, originService, destinationService);this.dataSync source;}/*** 事件的日志打印,会在监听器监听到事件时输出打印* 结果尽可能不要有换行,保证日志输出在一行内* 该方法可以在子类中重写** return 日志*/public String logPrint() {return String.format({\originService\:\%s\,\destinationService\:\%s\,\id\:\%s\,\dataSync\:%s,\timestamp\:\%s\}, this.getId(), this.getOriginService(), this.getDestinationService(), Objects.nonNull(this.dataSync) ? this.dataSync.toString() : null, this.getTimestamp());}/*** 数据同步的原始数据封装*/public static class DataSyncT implements Serializable {private DataSyncOperateTypeEnum operateType;private T data;public DataSync() {}public DataSync(DataSyncOperateTypeEnum operateType, T data) {this.operateType operateType;this.data data;}public DataSyncOperateTypeEnum getOperateType() {return operateType;}public T getData() {return data;}Overridepublic String toString() {return {\operateType\: operateType ,\data\: data };}}
}使用bus
引用注入bus应用的接口远程调用
定义bus远程调用
FeignClient(valuesms-bus, fallbackFactory DataSyncNotifyEventServiceFallbackFactory.class)
public interface DataSyncNotifyEventService {/*** 发布数据同步通知事件* destination 参数被删除,不需要指定服务** param eventEnum 事件枚举,可通过枚举找到对应的事件类* param operateType 操作类型枚举* param obj 需要处理的消息* return 发布结果*/PostMapping(/default/publish/{eventEnum}/{operateType})Result publishDataSyncNotifyEvent(PathVariable(eventEnum) DataSyncEventEnum eventEnum,PathVariable(operateType) DataSyncOperateTypeEnum operateType,RequestBody Object obj,RequestParam(destination) String destination);
}注入使用
A应用数据更新后通过bus数据同步给B应用
在A应用的业务层写以下代码 try {result dataSyncNotifyEventService.publishDataSyncNotifyEvent(DataSyncEventEnum.ROUTE_SYN,DataSyncOperateTypeEnum.ADD,new ArrayListRoute() {{Route r new Route();r.setCreateTime(date);add(r);}}, ServiceEnum.SMS_RULES.serviceName);log.info(新增路由调用通知同步所有服务 result:{}, result);} catch (Exception e) {log.error(同步异常 {}, result, e);}
