Zookeeper发布订阅之SpringBoot+Mybatis多数据源

   1.前言

　　　数据发布/订阅系统，即所谓的配置中心，顾名思义就是发布者将数据发布到Zookeeper的一个或一系列节点上，供订阅者进行数据订阅，进而达到动态获取数据的目的，实现配置信息的集中管理和数据的动态更新。

          发布订阅一般有两种模式，分别是推（Push）和拉（Pull）模式。在推拉模式中，服务端主动将数据更新发送给所有订阅的客户端;而拉模式则是客户端主动发起请求来获取最新数据。Zookeeper采取推和拉相结合的方式：客户端会向相应的客户端发送Watcher事件通知，客户端接收到这个消息通知后，需要主动到服务端获取最新的数据。

　　   “配置管理”的实际案例来展示Zookeeper在"数据发布/订阅"场景下的使用方式。

   2.案例

           将敏感的配置信息存放与Zookeeper中。启动项目时，自动从Zookeeper中获取。

          Maven坐标：

　　    

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.9.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.9.0</version>
        </dependency>
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis</artifactId>
            <version>3.4.3</version>
        </dependency>
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis-spring</artifactId>
            <version>1.3.1</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
            <version>2.7.4</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

         自动从Zookeeper中获取”配置信息“，并放入Enviroment中

             配置ApplicationContextInitializer，项目在启动先调用该类，进行加载环境变量到Enviroment中去（这里采用在classpath路径下新建META-INF/spring.factories文件方式，其它两种方式为：

　　　　1.使用SpringApplication 对象，调用 addInitializers（）

　　　　2.在配置文件中配置 context.initializer.classes =xx）

               spring.factories文件

 　　　　

# Initializers
org.springframework.context.ApplicationContextInitializer=
com.example.zkconfig.config.GlobalConfigInit

　　 　　

GlobalConfigIn类进行一系列配置文件加载及Zk连接，我用的Curator框架

我的application.properties此刻内容

server.port=8076
#zk 配置信息
com.unconfig.info=/config/zk-config

　application-dev 则对应具体连接机器

#zk
com.zookeeper.url=zk://192.168.159.129:2181

 从zk取配置信息放入spring环境

@Slf4j
public class GlobalConfigInit implements ApplicationContextInitializer<ConfigurableApplicationContext> {

    @Override
    public void initialize(ConfigurableApplicationContext configurableApplicationContext) {
        ConfigurableEnvironment environment = configurableApplicationContext.getEnvironment();
        String zkUri = environment.getProperty("com.zookeeper.url");
        if (null == zkUri) {
            log.warn("could not find zk uri,unconf not configured..");
        } else {
            ZookeeperUtils zookeeperUtils = new ZookeeperUtils();
            zookeeperUtils.setZkUri(zkUri);
            String unconf = environment.getProperty("com.unconfig.info");
            if (null != unconf) {
                log.info("config uniconf .. {}", unconf);
                ZkPropertySource propertySource = new ZkPropertySource("default", zookeeperUtils.getEnv(unconf));
                environment.getPropertySources().addLast(propertySource);
            } else {
                log.warn("unconf not configured..");
            }
        }
    }

    public String getEnv(Environment environment) {
        String zkUrl = environment.getProperty("com.zookeeper.url");
        if (null != zkUrl && !zkUrl.isEmpty()) {
            return zkUrl;
        } else {
            String unConfig = environment.getProperty("com.unconfig.info");
            if (unConfig != null && !unConfig.isEmpty()) {
                ZookeeperURI parse = ZookeeperURI.parse(unConfig);
                String inferZkUrl = "zk://" + parse.getServers();
                System.setProperty("com.zookeeper.url", inferZkUrl);
                log.info("inferred zk uri from unconf zk {}", inferZkUrl);
                return inferZkUrl;
            } else {
                return null;
            }
        }
    }
}

 　　　　这里不再多说，接下来配置数据源。

　　

@Configuration
@MapperScan(basePackages = {"com.example.zkconfig.dao.write"}, sqlSessionFactoryRef = "writeSqlSessionFactory")
public class WriteDataSourceConfig {

    @Value("${write.datasource.mappers}")
    private String location;

    @Bean(name = "writeDataSource")
    @ConfigurationProperties(prefix = "write.datasource")
    public DataSource dataSource() {
        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }

    @Bean(name = "writeSqlSessionFactory")
    public SqlSessionFactory writeSqlSessionFactory(@Qualifier("writeDataSource") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSource);
        sqlSessionFactoryBean.setMapperLocations(
                new PathMatchingResourcePatternResolver().getResources(location));
        return sqlSessionFactoryBean.getObject();
    }

    @Bean(name = "writeTransactionManager")
    public DataSourceTransactionManager writeTransactionManager(@Qualifier("writeDataSource") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }
}

  　　　　而我实际在zookeeper节点配置的数据

　　

　　　　

测试走一波，启动成功。