spring-bus消息总线的使用

文章目录

依赖
bus应用
- 接口
- 用到的封装参数类
接收的应用
- 监听器
- 定义的事件类
使用bus
- 定义bus远程调用
- A应用数据更新后通过bus数据同步给B应用

依赖

   <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bus-amqp</artifactId></dependency>

bus应用

类似于生产者

接口

供内部其他应用使用,远程调用该接口实现各应用之间数据同步
参数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 DataSyncEvent<T> extends RemoteApplicationEvent {/*** 事件数据*/private DataSync<T> dataSync;public DataSync<T> getDataSync() {return dataSync;}public void setDataSync(DataSync<T> dataSync) {this.dataSync = dataSync;}/*** 基础构造器** @param source             引发事件的原始数据* @param originService      引发事件的原始服务* @param destinationService 事件的目标服务*/public DataSyncEvent(DataSync<T> 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 DataSync<T> 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 DataSyncListener<RouteSyncEvent, List<Route>> {private static final Log log = LogFactory.get();@Autowiredprivate RouteMapper routeMapper;@Autowiredprivate SynService synService;@Overridepublic void handleByADD(List<Route> data) {Optional.ofNullable(data).ifPresent(ls -> {if (ls.isEmpty()) {return;}long l = System.currentTimeMillis();String time = DateUtil.formatDateTime(ls.get(0).getCreateTime());List<Route> 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(List<Route> data) {// 流程还是先删除后新增的方式Optional.ofNullable(data).ifPresent(ls -> {List<Integer> collect = ls.stream().map(Route::getId).filter(Objects::nonNull).collect(Collectors.toList());if (collect.isEmpty()) {return;}long l = System.currentTimeMillis();// 如果有，先删除List<Route> 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));}// 如果有，再加入List<Route> 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(List<Route> data) {Optional.ofNullable(data).ifPresent(ls -> {long l = System.currentTimeMillis();List<Route> 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 DataSyncListener<T extends DataSyncEvent<D>, D> implements ApplicationListener<T> {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.DataSync<D> 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 DataSyncEvent<List<Route>> {private static final long serialVersionUID = -501657066268464154L;public RouteSyncEvent(DataSyncOperateTypeEnum operateType, List<Route> Routes, String originService, String destinationService) {super(new DataSync<>(operateType, Routes), originService, destinationService);}
}

继承的抽象事件类

/*** 数据同步通知事件,作为一般通用事件使用,如需要特殊处理建议新增事件*/
public abstract class DataSyncEvent<T> extends RemoteApplicationEvent {/*** 事件数据*/private DataSync<T> dataSync;public DataSync<T> getDataSync() {return dataSync;}public void setDataSync(DataSync<T> dataSync) {this.dataSync = dataSync;}/*** 基础构造器** @param source             引发事件的原始数据* @param originService      引发事件的原始服务* @param destinationService 事件的目标服务*/public DataSyncEvent(DataSync<T> 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 DataSync<T> 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(value="sms-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 ArrayList<Route>() {{Route r = new Route();r.setCreateTime(date);add(r);}}, ServiceEnum.SMS_RULES.serviceName);log.info("新增路由调用通知同步所有服务 result:{}", result);} catch (Exception e) {log.error("同步异常 {}", result, e);}