Java動(dòng)態(tài)線程池插件dynamic-tp集成zookeeper怎么配置

這篇文章主要介紹了Java動(dòng)態(tài)線程池插件dynamic-tp集成zookeeper怎么配置的相關(guān)知識(shí)，內(nèi)容詳細(xì)易懂，操作簡(jiǎn)單快捷，具有一定借鑒價(jià)值，相信大家閱讀完這篇Java動(dòng)態(tài)線程池插件dynamic-tp集成zookeeper怎么配置文章都會(huì)有所收獲，下面我們一起來(lái)看看吧。

前言

dynamic-tp是一個(gè)輕量級(jí)的動(dòng)態(tài)線程池插件，它是一個(gè)基于配置中心的動(dòng)態(tài)線程池，線程池的參數(shù)可以通過(guò)配置中心配置進(jìn)行動(dòng)態(tài)的修改，在配置中心的支持上最開(kāi)始的時(shí)候支持Nacos和Apollo。

配置刷新

dynamic-tp提供了一個(gè)刷新配置的接口Refresher，抽象類(lèi)AbstractRefresher實(shí)現(xiàn)刷新配置接口的刷新配置方法refresh，該方法能根據(jù)配置類(lèi)型內(nèi)容和配置解析配置并刷新動(dòng)態(tài)線程池的相關(guān)配置，由DtpRegistry負(fù)責(zé)刷新線程池配置，事件發(fā)布訂閱模式操作Web容器參數(shù)，代碼如下：

public interface Refresher {
    /**
     * Refresh with specify content.
     * @param content content
     * @param fileType file type
     */
    void refresh(String content, ConfigFileTypeEnum fileType);
}

@Slf4j
public abstract class AbstractRefresher implements Refresher {
    @Resource
    private DtpProperties dtpProperties;
    @Resource
    private ApplicationEventMulticaster applicationEventMulticaster;
    @Override
    public void refresh(String content, ConfigFileTypeEnum fileTypeEnum) {
        if (StringUtils.isBlank(content) || Objects.isNull(fileTypeEnum)) {
            return;
        }
        try {
            // 根據(jù)配置內(nèi)容和配置類(lèi)型將配置內(nèi)容轉(zhuǎn)成Map
            val prop = ConfigHandler.getInstance().parseConfig(content, fileTypeEnum);
            doRefresh(prop);
        } catch (IOException e) {
            log.error("DynamicTp refresh error, content: {}, fileType: {}",
                    content, fileTypeEnum, e);
        }
    }
    private void doRefresh(Map<Object, Object> properties) {
        // 將Map中的配置轉(zhuǎn)換成DtpProperties
        ConfigurationPropertySource sources = new MapConfigurationPropertySource(properties);
        Binder binder = new Binder(sources);
        ResolvableType type = ResolvableType.forClass(DtpProperties.class);
        Bindable<?> target = Bindable.of(type).withExistingValue(dtpProperties);
        binder.bind(MAIN_PROPERTIES_PREFIX, target);
        // 刷新動(dòng)態(tài)線程池配置
        DtpRegistry.refresh(dtpProperties);
        // 發(fā)布刷新實(shí)現(xiàn)，該事件用于控制Web容器線程池參數(shù)控制
        publishEvent();
    }
    private void publishEvent() {
        RefreshEvent event = new RefreshEvent(this, dtpProperties);
        applicationEventMulticaster.multicastEvent(event);
    }
}

Zookeeper配置中心接入擴(kuò)展實(shí)現(xiàn)

基于AbstractRefresher就可以實(shí)現(xiàn)Zookeeper配置中心的擴(kuò)展了，Zookeeper的擴(kuò)展實(shí)現(xiàn)繼承AbstractRefresher，Zookeeper的擴(kuò)展實(shí)現(xiàn)只需要監(jiān)聽(tīng)配置中心的配置變更即可拿到配置內(nèi)容，然后通過(guò)refresh刷新配置即可。代碼如下：

ZookeeperRefresher繼承AbstractRefresher，實(shí)現(xiàn)InitializingBean，afterPropertiesSet方法邏輯從配置DtpProperties獲取Zookeeper的配置信息，CuratorFrameworkFactory創(chuàng)建客戶(hù)端，設(shè)置監(jiān)聽(tīng)器，這里有兩種監(jiān)聽(tīng)器，一個(gè)是連接監(jiān)聽(tīng)ConnectionStateListener，一個(gè)是節(jié)點(diǎn)變動(dòng)監(jiān)聽(tīng)CuratorListener，出發(fā)監(jiān)聽(tīng)后loadNode負(fù)責(zé)從Zookeeper獲取配置文件配置并組裝配置內(nèi)容，然后通過(guò)refresh刷新配置，注意，Zookeeper配置目前配置類(lèi)型僅支持properties。

@Slf4j
public class ZookeeperRefresher extends AbstractRefresher implements InitializingBean {
    @Resource
    private DtpProperties dtpProperties;
    private CuratorFramework curatorFramework;
    @Override
    public void afterPropertiesSet() throws Exception {
        DtpProperties.Zookeeper zookeeper = dtpProperties.getZookeeper();
        curatorFramework = CuratorFrameworkFactory.newClient(zookeeper.getZkConnectStr(),
                new ExponentialBackoffRetry(1000, 3));
        String nodePath = ZKPaths.makePath(ZKPaths.makePath(zookeeper.getRootNode(),
                zookeeper.getConfigVersion()), zookeeper.getNode());
        final ConnectionStateListener connectionStateListener = (client, newState) -> {
            if (newState == ConnectionState.CONNECTED || newState == ConnectionState.RECONNECTED) {
                loadNode(nodePath);
            }};
        final CuratorListener curatorListener = (client, curatorEvent) -> {
            final WatchedEvent watchedEvent = curatorEvent.getWatchedEvent();
            if (null != watchedEvent) {
                switch (watchedEvent.getType()) {
                    case NodeChildrenChanged:
                    case NodeDataChanged:
                        loadNode(nodePath);
                        break;
                    default:
                        break;
                }
            }};
        curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
        curatorFramework.getCuratorListenable().addListener(curatorListener);
        curatorFramework.start();
        log.info("DynamicTp refresher, add listener success, nodePath: {}", nodePath);
    }
    /**
     * load config and refresh
     * @param nodePath config path
     */
    public void loadNode(String nodePath) {
        try {
            final GetChildrenBuilder childrenBuilder = curatorFramework.getChildren();
            final List<String> children = childrenBuilder.watched().forPath(nodePath);
            StringBuilder content = new StringBuilder();
            children.forEach(c -> {
                String n = ZKPaths.makePath(nodePath, c);
                final String nodeName = ZKPaths.getNodeFromPath(n);
                final GetDataBuilder data = curatorFramework.getData();
                String value = "";
                try {
                    value = new String(data.watched().forPath(n), StandardCharsets.UTF_8);
                } catch (Exception e) {
                    log.error("zk config value watched exception.", e);
                }
                content.append(nodeName).append("=").append(value).append("\n");
            });
            refresh(content.toString(), ConfigFileTypeEnum.PROPERTIES);
        } catch (Exception e) {
            log.error("load zk node error, nodePath is {}", nodePath, e);
        }
    }
}

關(guān)于“Java動(dòng)態(tài)線程池插件dynamic-tp集成zookeeper怎么配置”這篇文章的內(nèi)容就介紹到這里，感謝各位的閱讀！相信大家對(duì)“Java動(dòng)態(tài)線程池插件dynamic-tp集成zookeeper怎么配置”知識(shí)都有一定的了解，大家如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道。