SpringBoot+ZooKeeper+ZKUI+Drools 实现应用配置中心及业务规则动态加载

本文目的：

使用ZooKeeper作为SpringBoot应用的配置中心

应用中使用到的业务规则存储在Zookeeper中，规则更新后在不重启应用的情况下通知应用动态重载规则

1.zookeeper简介

Zookeeper是一个高性能，分布式的，开源分布式应用协调服务。它提供了简单原始的功能，分布式应用可以基于它实现更高级的服务，比如同步，配置管理，集群管理，命名空间。它被设计为易于编程，使用文件系统目录树作为数据模型。服务端跑在java上，并且提供java和C的客户端API。

数据模型：



特点：

采用树形结构，每个节点叫Znode，节点路径已/分隔，如：/zoo/foo,每个节点路径必须唯一，且没有相对路径

每个Znode都可以存储数据，数据类型byte[]，都可以有子节点

每个Znode都有一个stat数据结构来存储数据的版本，ACL及时间戳等

每个Znode都可以进行CRWD操作

由此可见，采用zookeeper可以很方便的管理应用的配置，如：名为cyzy-gpserver的springboot应用，可以将配置存储在：

/cyzy-gpserver ## 等同 application.yml

/cyzy-gpserver,dev ## 等同 application-dev.yml

/cyzy-gpserver,prod ## 等同 application-prod.yml

这些节点下，每个节点下再分别存储相关配置项，如：

2.zookeeper及其ui客户端安装

zookeeper：

前往http://www.apache.org/dyn/closer.cgi/zookeeper/，下载3.4.6

修改如下配置，zookeeper-3.4.6\conf\zoo.cfg:

# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=D:\dev\zookeeper-3.4.6\datas

运行：

windows：zookeeper-3.4.6\bin\zkServer.cmd

linux：zookeeper-3.4.6\bin\zkServer.sh

ui客户端zkui：

前往https://github.com/DeemOpen/zkui，参考文档自行编译打包。

或使用别人做好的

http://download.csdn.net/detail/lirenzuo/9640272

登陆用户名/密码配置，zkui\config.cfg:

userSet = {"users": [{ "username":"admin" , "password":"admin","role": "ADMIN" }

运行：

java -jar xxx.jar

3.与SpringBoot整合

pom：

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.4.0.RELEASE</version>
</parent>

<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-zookeeper-config</artifactId>
</dependency>
</dependencies>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>Camden.SR2</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>

src\main\resources下的bootstrap.yml：

spring:
application:
name: cyzy-gpserver
profiles:
active: prod
cloud:
zookeeper:
enabled: true  # true:开启zookeeper外部化配置, false:读取本地配置; 需要将config.enabled,config.watcher.enabled同时设置
connect-string: 127.0.0.1:2181
config:
enabled: true
watcher:
enabled: false

这样默认情况下应用会去/config/application,prod 和 /config/cyzy-gpserver,prod节点下读取配置

注意：/config/application节点下的配置项会应用到所有应用

具体参考：

http://cloud.spring.io/spring-cloud-static/spring-cloud-zookeeper/1.0.3.RELEASE/#spring-cloud-zookeeper-config

4.业务规则动态加载

基本思路:

1.规则存储在zookeeper中，按照不同的Znode进行分组，如：某类规则存储在/drools.rules/group1下, 某类存储在/drools.rules/group2

2.每个存储了规则的Znode下创建一个“状态节点”和一个“结果反馈节点”，如：

/drools.rules/group1/a_push_node, /drools.rules/group1/a_result_node

3.应用使用curator客户端的TreeCacheListener方式来监控/drools.rules下所有”状态节点”的变化，根据其状态值，进行相应操作。如：在/drools.rules/group1下修改了某个规则节点，此时设置a_push_node值为0,则应用重载/drools.rules/group1下的规则进行编译测试；设置为1，应用重载规则。应用操作结果反馈回result_node。

4.应用在使用时可以按照Znode的路径来fire rules，如：fireRules(String ruleInZKPath, String agendaGroup, AgendaFilter filter, Object… facts)

这样就可以实现一个轻量级的简易的规则管理和动态重载规则的系统……

整合Drools：

<dependency>
<groupId>org.drools</groupId>
<artifactId>drools-core</artifactId>
<version>6.4.0.Final</version>
</dependency>
<dependency>
<groupId>org.drools</groupId>
<artifactId>drools-compiler</artifactId>
<version>6.4.0.Final</version>
</dependency>

动态加载规则服务：

/**
* 可动态加载Drools规则的服务，加载方式：
* 1.从本地文件加载，通过http get：/app/drools/reload 或 /app/drools/test 重载规则
* 2.从远程zookeeper节点上加载，通过设置节点下a_push_node值的方式重载规则：0，测试模式；非0，正式加载
* @ClassName: DroolsService
*/
@Service
@RestController
public class DroolsService{
private static final Logger LOG = LoggerFactory.getLogger(DroolsService.class);

private static final String GROUP_NAME = "com.cyzy";
private static fina
4000
l String VERSION = "1.0.0";
private static final String ZK_ENABLED = "spring.cloud.zookeeper.enabled";
private static final String ZK_PRIFEX_NODE_PATH = "/drools.rules.test";
private static final String ZK_NODE_TEST_VALUE = "0";
private static final String PRIFEX_DIR="drools.prefixDir";
private static final String TARGET_DIR="drools.targetDir";
private static final String RELOAD_RULES_OK = "reload drools's rules ok......";

@Autowired
private Environment env;

@Autowired
private ApplicationContext appCtx;

private CuratorFramework client;

private KieServices kieServices;

private KieRepository repository;

private ConcurrentHashMap<String, KieContainer> kieContainers = new ConcurrentHashMap<String, KieContainer>();

@PostConstruct
protected void initKieContainer() throws Exception {
kieServices = KieServices.Factory.get();
repository = kieServices.getRepository();

boolean isZKEnabled = env.getProperty(ZK_ENABLED, Boolean.class);
if (isZKEnabled) {
client = appCtx.getBean(CuratorFramework.class);
loadRulesFromZK();
} else {
loadConfigFromLocalFile();
}
}

/**
* 从ZK加载规则
* @Title: loadRulesFromZK
* @throws Exception
* @return: void
*/
private void loadRulesFromZK() throws Exception{
ZkNodeListenerAdapter.getInstance()
//设置watcher
.watcher(new ZKNodeWatcher(ZK_PRIFEX_NODE_PATH, client))
//默认处理器
.setDefaultHandler(new ZkNodeHandler() {
@Override
public String execute(CuratorFramework client, TreeCache tc, TreeCacheEvent event, ChildData targetChildData,
Map<String, ChildData> filterChildren) {
String msg = RELOAD_RULES_OK;
try {
String pathName = genPathName(event.getData().getPath());
loadRules(pathName, getRulesFromZKNodes(pathName, filterChildren));
} catch (Exception e) {
msg = e.getMessage();
LOG.error(msg);
}
return msg;
}
})
//a_push_node节点值为0时的处理器
.addHandler(ZK_NODE_TEST_VALUE, new ZkNodeHandler() {
@Override
public String execute(CuratorFramework client, TreeCache tc, TreeCacheEvent event, ChildData targetChildData,
Map<String, ChildData> filterChildren) {
String msg = RELOAD_RULES_OK;
try {
String pathName = genPathName(event.getData().getPath());
testRules(pathName, getRulesFromZKNodes(pathName, filterChildren));
} catch (Exception e) {
msg = e.getMessage();
LOG.error(msg);
}
return msg;
}
})
//增加此监听器到watcher
.addToWatcher(ZK_PRIFEX_NODE_PATH+"/gps/zte")
.addToWatcher(ZK_PRIFEX_NODE_PATH+"/gps/zte2")
.addToWatcher(ZK_PRIFEX_NODE_PATH+"/group1")
.addToWatcher(ZK_PRIFEX_NODE_PATH+"/group2")
//启动监听
.start();
}

/**
* 从本地文件加载规则
* @Title: loadConfigFromLocalFile
* @throws Exception
* @return: void
*/
private void loadConfigFromLocalFile() throws Exception{
String targetDir = env.getProperty(TARGET_DIR);
loadRules(targetDir, getRulesFromLocalFile(targetDir));
}

/**
* 正式加载规则文件进行使用
*/
private void loadRules(String ruleInZKPath, List<ResourceWrapper> resourceWrappers) throws Exception {
// if failed throws Exception
ReleaseId releaseId = kieServices.newReleaseId(GROUP_NAME, ruleInZKPath, VERSION);
InternalKieModule kieModule = DroolsUtils.createKieModule(kieServices, releaseId, resourceWrappers);
// if succeed will add new module
repository.addKieModule(kieModule);
KieContainer kieContainer = kieServices.newKieContainer(releaseId);
kieContainer.updateToVersion(releaseId);
kieContainers.put(ruleInZKPath, kieContainer);
}

/**
* 测试加载规则文件
*/
private void testRules(String ruleInZKPath, List<ResourceWrapper> resourceWrappers) throws Exception{
ReleaseId releaseId = kieServices.newReleaseId(GROUP_NAME, ruleInZKPath, VERSION);
DroolsUtils.createKieModule(kieServices, releaseId, resourceWrappers);
}

/**
* 从ENV配置的指定文件夹中获取规则文件
*/
private List<ResourceWrapper> getRulesFromLocalFile(String path) {
String prefixDir = env.getProperty(PRIFEX_DIR);
String targetDir = env.getProperty(TARGET_DIR);
List<ResourceWrapper> resourceWrappers = new ArrayList<ResourceWrapper>();
List<File> ruleFiles = new ArrayList<File>();
try {
ruleFiles = DroolsUtils.getDefaultRuleFiles(prefixDir, targetDir);
} catch (Exception e) {}
if(ruleFiles.isEmpty()){
throw new RuntimeException("can't load rules from " + prefixDir +"/"+ targetDir);
}
LOG.info("################################load ["+path+"] rules################################");
for (File file : ruleFiles) {
LOG.info(file.getName());
resourceWrappers.add(new ResourceWrapper(ResourceFactory.newFileResource(file), file.getName()));
}
LOG.info("################################load ["+path+"] rules################################");
return resourceWrappers;
}

/**
* 从ZK节点上获取规则文件
*/
private List<ResourceWrapper> getRulesFromZKNodes(String path, Map<String, ChildData> filterChildren) {
List<ResourceWrapper> resourceWrappers = new ArrayList<ResourceWrapper>();
LOG.info("################################load ["+path+"] rules################################");
for (Entry<String, ChildData> entry : filterChildren.entrySet()) {
LOG.info(entry.getKey());
resourceWrappers.add(
new ResourceWrapper(ResourceFactory.newByteArrayResource(entry.getValue().getData()),entry.getKey()));
}
LOG.info("################################load ["+path+"] rules################################");
return resourceWrappers;
}

private String genPathName(String zkNodePath) {
String name = "";
Matcher matcher = Pattern.compile("(.*?)/a_push_node").matcher(zkNodePath);
if (matcher.matches()) {
name = matcher.group(1);
}
return name;
}

/**
* 创建指定ZK节点路径下的KieSession
* @Title: newSession
* @Description: TODO
* @param ruleInZKPath
* @return: KieSession
*/
private KieSession newSession(String ruleInZKPath) {
KieContainer kieContainer = kieContainers.get(ruleInZKPath);
if(kieContainer==null){
throw new RuntimeException("can't get KieContainer with the name:" + ruleInZKPath);
}
KieSession session = kieContainer.newKieSession();
// 默认配置
session.setGlobal("appCtx", appCtx);
return session;
}

/**
* 执行指定ZK节点路径下的所有“MAIN”议程组(使用agenda-group定义，默认是MAIN)的规则
* @Title: fireAllRules
* @param ruleInZKPath
* @param facts
* @return: void
*/
public void fireMainGroupRules(String ruleInZKPath, Object... facts) {
fireRules(ruleInZKPath, null, null, facts);
}

/**
* 执行指定ZK节点路径下的所有“MAIN”议程组(使用agenda-group定义，默认是MAIN)的并且经过AgendaFilter过滤的规则
* @Title: fireAllRules
* @param ruleInZKPath
* @param filter
* @param facts
* @return: void
*/
public void fireMainGroupRules(String ruleInZKPath, AgendaFilter filter, Object... facts) {
fireRules(ruleInZKPath, null, filter, facts);
}

/**
* 执行指定ZK节点路径下的所有agendaGroup指定议程组和“MAIN”议程组(使用agenda-group定义，默认是MAIN)的并且经过AgendaFilter过滤的规则
* @Title: fireAllRules
* @param ruleInZKPath 规则所在ZK node的路径
* @param agendaGroup 规则所在议程组名称，如果不定义默认：MAIN
* @param filter 规则过滤器
* @param facts 事实
* @return: void
*/
public void fireRules(String ruleInZKPath, String agendaGroup, AgendaFilter filter, Object... facts) {
KieSession session = null;
try {
session = newSession(ruleInZKPath);
// add fact
for (Object fact : facts) {
session.insert(fact);
}
//focus agenda group
if (agendaGroup != null && !agendaGroup.isEmpty()) {
session.getAgenda().getAgendaGroup(agendaGroup).setFocus();
}
// add filter
if (filter != null) {
session.fireAllRules(filter);
} else {
session.fireAllRules();
}
} catch (Exception e) {
throw new RuntimeException(e.getMessage());
} finally {
if (session != null) {
session.dispose();
}
}
}

@RequestMapping(value = "/app/drools/reload", method = RequestMethod.GET)
public @ResponseBody ResponseEntity<String> reloadRules() {
String msg = RELOAD_RULES_OK;
try {
String targetDir = env.getProperty(TARGET_DIR);
loadRules(targetDir, getRulesFromLocalFile(targetDir));
} catch (Exception e) {
msg = e.getMessage();
}
return ResponseEntity.ok().body(msg);
}

@RequestMapping(value = "/app/drools/test", method = RequestMethod.GET)
public @ResponseBody ResponseEntity<String> testRules() {
String msg = RELOAD_RULES_OK;
try {
String targetDir = env.getProperty(TARGET_DIR);
testRules(targetDir, getRulesFromLocalFile(targetDir));
} catch (Exception e) {
msg = e.getMessage();
}
return ResponseEntity.ok().body(msg);
}

@RequestMapping(value = "/app/drools/test2", method = RequestMethod.GET)
public @ResponseBody ResponseEntity<String> test2() {
String msg = "test ok";
try {
fireRules(ZK_PRIFEX_NODE_PATH + "/group1", "group1", null, new Object());
fireRules(ZK_PRIFEX_NODE_PATH + "/group2", "", null, new Object());

fireRules(ZK_PRIFEX_NODE_PATH + "/gps/zte", "gps.zte", null, new Object());
fireRules(ZK_PRIFEX_NODE_PATH + "/gps/zte2", "gps.zte", null, new Object());

//          fireAllRules("rules", null, null, new Object());
} catch (Exception e) {
msg = e.getMessage();
}
return ResponseEntity.ok().body(msg);

}
}