本文介绍在JAVA应用中如何连接consul并使用consul作为K-V存储中心。

consul控制台.png

consul作为注册中心，但是也提供了kv的存储功能。

1. 引入依赖

        <dependency>
            <groupId>com.ecwid.consul</groupId>
            <artifactId>consul-api</artifactId>
            <version>1.4.5</version>
        </dependency>

开始的使用引入com.orbitz.consul依赖，但是请求失败，最终选择了上面的pom依赖。

2. API使用

存取实现

    public static void main(String[] args) {
        ConsulClient leaderConsul = new ConsulClient("127.0.0.1", 8500);
        //存入k-v
        leaderConsul.setKVValue("name1","hah");
        //取出k-v
        Response<GetValue> kvValue = leaderConsul.getKVValue("name1");
        //输出的是解码后的文件
        System.out.println(kvValue.getValue().getDecodedValue());
    }

存取目录格式

consul和zk相同，当存储的key以/分割时，显示是目录的形式。

需求：consul中存储数据的格式为：appRules/项目名/操作名，获取所有的目录名：

    private Set<String> getAppNameFromConsul() {
        Response<List<String>> appRules = consulClient.getKVKeysOnly("appRules");
        List<String> results = appRules.getValue();
        //规则的截取-appRules/项目名/规则名
        Set<String> consulSet = new HashSet<>();
        if (results != null) {
            for (String result : results) {
                try {
                    String[] arr = result.split("/");
                    consulSet.add(arr[1]);
                } catch (Exception e) {
                    logger.error("", e);
                }
            }
        }
        return consulSet;
    }

使用如上的截取方式，得到的对应的项目名。

实现分布式锁

锁的接口：

public interface Lock {

    String lock(String lockName, int ttlSeconds);

    void unLock(String sessionID);

    /**
     * 线程安全的操作
     *
     * @param lockName   锁名称
     * @param ttlSeconds 最大超时时间
     * @param supplier   待保护的临界资源
     * @param <T>        返回值的类型
     * @return 临界资源返回值
     */
    default <T> T threadSafetyFunction(String lockName, int ttlSeconds, Supplier<T> supplier) {
        T result;
        String successId = lock(lockName, 10);
        if (successId != null) {
            try {
                result = supplier.get();
            } finally {
                unLock(successId);
            }
        } else {
            throw new RuntimeException("冲突啦，请稍等处理~");
        }
        return result;

    }
}

consul分布式锁实现类：

public class ConsulLock implements Lock {

    private ConsulClient consulClient;

    /**
     * 构造函数
     *
     * @param consulHost 注册consul的client或服务端的Ip或主机名，或域名
     * @param consulPort 端口号
     */
    public ConsulLock(String consulHost, int consulPort) {
        consulClient = new ConsulClient(consulHost, consulPort);
    }


    public ConsulLock(ConsulClient consulClient) {
        this.consulClient = consulClient;
    }

    /**
     * 获得锁的方法
     *
     * @param lockName   竞争的资源名
     * @param ttlSeconds 锁的超时时间，超过该时间自动释放【超时时间,最低10秒】
     * @return 若为null表示加锁失败，若非null，表示加锁成功
     */
    public String lock(String lockName, int ttlSeconds) {
        if (ttlSeconds < 10 || ttlSeconds > 86400) ttlSeconds = 10;
        String sessionId = createSession(lockName, ttlSeconds);
        //个性化处理
        boolean success = lock("sentinel/lock/" + lockName, sessionId);
        if (!success) {
            consulClient.sessionDestroy(sessionId, null);
            return null;
        }
        return sessionId;
    }

    public void unLock(String sessionID) {
        consulClient.sessionDestroy(sessionID, null);
    }

    private String createSession(String lockName, int ttlSeconds) {
        NewCheck check = new NewCheck();
        check.setId("check " + lockName);
        check.setName(check.getId());
        check.setTtl(ttlSeconds + "s"); //该值和session ttl共同决定决定锁定时长
        check.setTimeout("10s");
        consulClient.agentCheckRegister(check);
        consulClient.agentCheckPass(check.getId());

        NewSession session = new NewSession();
        session.setBehavior(Session.Behavior.RELEASE);
        session.setName("session " + lockName);
        session.setLockDelay(1);
        session.setTtl(ttlSeconds + "s"); //和check ttl共同决定锁时长
        List<String> checks = new ArrayList<>();
        checks.add(check.getId());
        session.setChecks(checks);
        return consulClient.sessionCreate(session, null).getValue();
    }


    private boolean lock(String lockName, String sessionId) {
        PutParams putParams = new PutParams();
        putParams.setAcquireSession(sessionId);
        return consulClient.setKVValue(lockName, "lock:" + LocalDateTime.now(), putParams).getValue();
    }

}

相关阅读

consul的API文档