zianxin/pigx
1
0 Fork 0
mirror of http://git.pig4cloud.com/pig/pigx.git synced 2026-11-14 06:09:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(pigx-mp-platform): 重构 WxMpInitConfigRunner 类，添加详细文档注释

Browse Source 
- 优化消息路由规则配置，支持多种微信事件处理
- 增加 Redis 监听机制，支持运行时动态重新加载配置
- 完善多租户支持，缓存 appid 与租户的映射关系
This commit is contained in:
PIGCLOUD committed 2026-03-08 15:00:56 +08:00
1 parent 7e78541763
commit c19c0f1ed4
1 file changed
+229 -145
Download Patch File Download Diff File Expand all files Collapse all files
pigx-visual/pigx-mp-platform/src/main/java/com/pig4cloud/pigx/mp/config/WxMpInitConfigRunner.java
+229 -145
View File
@@ -33,66 +33,131 @@ import java.util.stream.Collectors;
import static me.chanjar.weixin.common.api.WxConsts.*; import static me.chanjar.weixin.common.api.WxConsts.*;
/** /**
* 微信公众号初始化配置类
* <p>
* 功能说明：
* 1. 应用启动时自动加载所有租户的微信公众号配置
* 2. 为每个公众号创建独立的 WxMpService 和消息路由器
* 3. 监听 Redis 消息，支持运行时动态重新加载配置
* <p>
* 核心数据结构：
* - routers: appid -> WxMpMessageRouter 映射，用于消息路由分发
* - mpServices: appid -> WxMpService 映射，用于调用微信API
* - tenants: appid -> tenantId 映射，用于多租户隔离
*
* @author Binary Wang * @author Binary Wang
* @author lengleng * @author lengleng
* <p>
* 微信公众号基础配置，初始化配置
*/ */
@Slf4j @Slf4j
@Configuration @Configuration
@RequiredArgsConstructor @RequiredArgsConstructor
public class WxMpInitConfigRunner { public class WxMpInitConfigRunner {
/** /**
* 保存 appid-router 的对应关系 * 消息路由器缓存
*/ * key: 公众号 appid
private static Map<String, WxMpMessageRouter> routers = Maps.newHashMap(); * value: 对应的消息路由器，负责将不同类型的消息分发到对应的处理器
*/
private static Map<String, WxMpMessageRouter> routers = Maps.newHashMap();
/** /**
* 保存 appid-mpservice 的对应关系 * 微信公众号服务缓存
*/ * key: 公众号 appid
private static Map<String, WxMpService> mpServices = Maps.newHashMap(); * value: 微信公众号服务实例，封装了微信API调用能力
*/
private static Map<String, WxMpService> mpServices = Maps.newHashMap();
/** /**
* 保存 appid-tenantId 的对应关系 * 租户映射缓存
*/ * key: 公众号 appid
private static final Map<String, Long> tenants = Maps.newHashMap(); * value: 租户ID，用于多租户数据隔离
*/
private static final Map<String, Long> tenants = Maps.newHashMap();
private final RemoteTenantService tenantService; /**
* 远程租户服务，用于获取所有租户信息
*/
private final RemoteTenantService tenantService;
private final WxAccountService accountService; /**
* 微信公众号账户服务，用于查询公众号配置
*/
private final WxAccountService accountService;
private final LogHandler logHandler; /**
* 日志处理器，记录所有消息事件日志（异步执行）
*/
private final LogHandler logHandler;
private final NullHandler nullHandler; /**
* 空处理器，用于不需要特殊处理的事件（如点击菜单链接）
*/
private final NullHandler nullHandler;
private final KfSessionHandler kfSessionHandler; /**
* 客服会话处理器，处理客服会话创建/关闭/转接事件
*/
private final KfSessionHandler kfSessionHandler;
private final StoreCheckNotifyHandler storeCheckNotifyHandler; /**
* 门店审核通知处理器，处理门店审核结果事件
*/
private final StoreCheckNotifyHandler storeCheckNotifyHandler;
private final LocationHandler locationHandler; /**
* 地理位置处理器，处理用户上报的地理位置信息
*/
private final LocationHandler locationHandler;
private final MenuHandler menuHandler; /**
* 菜单点击处理器，处理自定义菜单点击事件
*/
private final MenuHandler menuHandler;
private final MsgHandler msgHandler; /**
* 消息处理器，处理用户发送的普通消息（默认处理器）
*/
private final MsgHandler msgHandler;
private final UnsubscribeHandler unsubscribeHandler; /**
* 取消关注处理器，处理用户取消关注公众号事件
*/
private final UnsubscribeHandler unsubscribeHandler;
private final SubscribeHandler subscribeHandler; /**
* 关注处理器，处理用户关注公众号事件
*/
private final SubscribeHandler subscribeHandler;
private final ScanHandler scanHandler; /**
* 扫码处理器，处理用户扫描二维码事件
*/
private final ScanHandler scanHandler;
@Async /**
@Order * 应用启动完成事件监听
@EventListener({ ApplicationReadyEvent.class }) * 异步初始化所有租户的微信公众号服务
public void WebServerInit() { */
this.initServices(); @Async
} @Order
@EventListener({ApplicationReadyEvent.class})
public void WebServerInit() {
this.initServices();
}
public void initServices() { /**
// 获取全部租户 遍历所有租户对应的公众号列表 * 初始化微信公众号服务
List<WxAccount> accountList = new ArrayList<>(); * <p>
// @formatter:off * 执行流程：
* 1. 获取系统中所有租户
* 2. 遍历每个租户，查询其配置的公众号列表
* 3. 为每个公众号创建 WxMpService 和消息路由器
* 4. 缓存 appid 与租户的映射关系
*/
public void initServices() {
// 获取全部租户 遍历所有租户对应的公众号列表
List<WxAccount> accountList = new ArrayList<>();
// @formatter:off
RetOps.of(tenantService.list()) RetOps.of(tenantService.list())
.getData() .getData()
.orElseGet(Collections::emptyList) .orElseGet(Collections::emptyList)
@@ -101,133 +166,152 @@ public class WxMpInitConfigRunner {
); );
// @formatter:on // @formatter:on
mpServices = accountList.stream().map(a -> { mpServices = accountList.stream().map(a -> {
WxMpInRedisConfigStorage configStorage = new WxMpInRedisConfigStorage(); WxMpInRedisConfigStorage configStorage = new WxMpInRedisConfigStorage();
configStorage.setAppId(a.getAppid()); configStorage.setAppId(a.getAppid());
configStorage.setSecret(a.getAppsecret()); configStorage.setSecret(a.getAppsecret());
configStorage.setToken(a.getToken()); configStorage.setToken(a.getToken());
configStorage.setAesKey(a.getAeskey()); configStorage.setAesKey(a.getAeskey());
WxMpService service = new WxMpServiceImpl(); WxMpService service = new WxMpServiceImpl();
service.setWxMpConfigStorage(configStorage); service.setWxMpConfigStorage(configStorage);
routers.put(a.getAppid(), this.newRouter(service)); routers.put(a.getAppid(), this.newRouter(service));
tenants.put(a.getAppid(), a.getTenantId()); tenants.put(a.getAppid(), a.getTenantId());
return service; return service;
}).collect(Collectors.toMap(s -> s.getWxMpConfigStorage().getAppId(), a -> a, (existing, replace) -> existing)); }).collect(Collectors.toMap(s -> s.getWxMpConfigStorage().getAppId(), a -> a, (existing, replace) -> existing));
} }
private WxMpMessageRouter newRouter(WxMpService wxMpService) { /**
final WxMpMessageRouter newRouter = new WxMpMessageRouter(wxMpService); * 创建微信消息路由器
* <p>
* 配置消息路由规则，将不同类型的消息/事件分发到对应的处理器：
* - 客服会话事件 -> kfSessionHandler
* - 门店审核事件 -> storeCheckNotifyHandler
* - 菜单点击事件 -> menuHandler
* - 关注/取消关注事件 -> subscribeHandler/unsubscribeHandler
* - 地理位置事件 -> locationHandler
* - 扫码事件 -> scanHandler
* - 其他消息 -> msgHandler（默认处理器）
*
* @param wxMpService 微信公众号服务实例
* @return 配置完成的消息路由器
*/
private WxMpMessageRouter newRouter(WxMpService wxMpService) {
final WxMpMessageRouter newRouter = new WxMpMessageRouter(wxMpService);
// 记录所有事件的日志 （异步执行） #5637 大数据量瓶颈 // 记录所有事件的日志 （异步执行） #5637 大数据量瓶颈
newRouter.rule().handler(this.logHandler).next(); newRouter.rule().handler(this.logHandler).next();
// 接收客服会话管理事件 // 接收客服会话管理事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(WxMpEventConstants.CustomerService.KF_CREATE_SESSION) .event(WxMpEventConstants.CustomerService.KF_CREATE_SESSION)
.handler(this.kfSessionHandler) .handler(this.kfSessionHandler)
.end(); .end();
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(WxMpEventConstants.CustomerService.KF_CLOSE_SESSION) .event(WxMpEventConstants.CustomerService.KF_CLOSE_SESSION)
.handler(this.kfSessionHandler) .handler(this.kfSessionHandler)
.end(); .end();
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(WxMpEventConstants.CustomerService.KF_SWITCH_SESSION) .event(WxMpEventConstants.CustomerService.KF_SWITCH_SESSION)
.handler(this.kfSessionHandler) .handler(this.kfSessionHandler)
.end(); .end();
// 门店审核事件 // 门店审核事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(WxMpEventConstants.POI_CHECK_NOTIFY) .event(WxMpEventConstants.POI_CHECK_NOTIFY)
.handler(this.storeCheckNotifyHandler) .handler(this.storeCheckNotifyHandler)
.end(); .end();
// 自定义菜单事件 // 自定义菜单事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(MenuButtonType.CLICK) .event(MenuButtonType.CLICK)
.handler(this.menuHandler) .handler(this.menuHandler)
.end(); .end();
// 点击菜单连接事件 // 点击菜单连接事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(MenuButtonType.VIEW) .event(MenuButtonType.VIEW)
.handler(this.nullHandler) .handler(this.nullHandler)
.end(); .end();
// 关注事件 // 关注事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(EventType.SUBSCRIBE) .event(EventType.SUBSCRIBE)
.handler(this.subscribeHandler) .handler(this.subscribeHandler)
.end(); .end();
// 取消关注事件 // 取消关注事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(EventType.UNSUBSCRIBE) .event(EventType.UNSUBSCRIBE)
.handler(this.unsubscribeHandler) .handler(this.unsubscribeHandler)
.end(); .end();
// 上报地理位置事件 // 上报地理位置事件
newRouter.rule() newRouter.rule()
.async(false) .async(false)
.msgType(XmlMsgType.EVENT) .msgType(XmlMsgType.EVENT)
.event(EventType.LOCATION) .event(EventType.LOCATION)
.handler(this.locationHandler) .handler(this.locationHandler)
.end(); .end();
// 接收地理位置消息 // 接收地理位置消息
newRouter.rule().async(false).msgType(XmlMsgType.LOCATION).handler(this.locationHandler).end(); newRouter.rule().async(false).msgType(XmlMsgType.LOCATION).handler(this.locationHandler).end();
// 扫码事件 // 扫码事件
newRouter.rule().async(false).msgType(XmlMsgType.EVENT).event(EventType.SCAN).handler(this.scanHandler).end(); newRouter.rule().async(false).msgType(XmlMsgType.EVENT).event(EventType.SCAN).handler(this.scanHandler).end();
// 默认 // 默认
newRouter.rule().async(false).handler(this.msgHandler).end(); newRouter.rule().async(false).handler(this.msgHandler).end();
return newRouter; return newRouter;
} }
/** /**
* redis 监听配置,监听 mp_redis_reload_topic,重新加载配置 * 配置 Redis 消息监听容器
* @param redisConnectionFactory redis 配置 * <p>
* @return * 监听 mp_redis_reload_topic 频道，当收到消息时重新加载公众号配置
*/ * 适用场景：后台修改公众号配置后，通过发布 Redis 消息触发配置热更新
@Bean *
public RedisMessageListenerContainer redisContainer(RedisConnectionFactory redisConnectionFactory) { * @param redisConnectionFactory Redis 连接工厂
RedisMessageListenerContainer container = new RedisMessageListenerContainer(); * @return Redis 消息监听容器
container.setConnectionFactory(redisConnectionFactory); */
container.addMessageListener((message, bytes) -> { @Bean
log.warn("接收到重新加载公众号配置事件"); public RedisMessageListenerContainer redisContainer(RedisConnectionFactory redisConnectionFactory) {
initServices(); RedisMessageListenerContainer container = new RedisMessageListenerContainer();
}, new ChannelTopic(CacheConstants.MP_REDIS_RELOAD_TOPIC)); container.setConnectionFactory(redisConnectionFactory);
return container; container.addMessageListener((message, bytes) -> {
} log.warn("接收到重新加载公众号配置事件");
initServices();
}, new ChannelTopic(CacheConstants.MP_REDIS_RELOAD_TOPIC));
return container;
}
public static Map<String, Long> getTenants() { public static Map<String, Long> getTenants() {
return tenants; return tenants;
} }
public static Map<String, WxMpMessageRouter> getRouters() { public static Map<String, WxMpMessageRouter> getRouters() {
return routers; return routers;
} }
public static Map<String, WxMpService> getMpServices() { public static Map<String, WxMpService> getMpServices() {
return mpServices; return mpServices;
} }
} }