Spring Boot—20Zookeeper

https://docs.spring.io/spring-boot/docs/2.0.1.RELEASE/reference/htmlsingle/

pom.xml

<dependency>

    <groupId>mysql</groupId>

    <artifactId>mysql-connector-java</artifactId>

    <version>8.0.11</version>

</dependency>

<dependency>

    <groupId>org.apache.curator</groupId>

    <artifactId>curator-framework</artifactId>

    <version>2.12.0</version>

</dependency>

<dependency>

    <groupId>org.apache.curator</groupId>

    <artifactId>curator-x-discovery</artifactId>

    <version>2.12.0</version>

</dependency>

<dependency>

    <groupId>org.apache.curator</groupId>

    <artifactId>curator-x-discovery-server</artifactId>

    <version>2.12.0</version>

</dependency>

application.properties

#

server.address=0.0.0.0

server.port=8080

server.servlet.context-path=/test

server.session.timeout=300

server.error.path=/error

#

server.tomcat.accesslog.enabled=true

server.tomcat.accesslog.buffered=true

server.tomcat.accesslog.directory=D:/Project/JavaWeb/SpringBoot/04JPASpringBoot/logs

#

spring.jackson.date-format=yyyy-MM-dd HH:mm:ss

spring.jackson.time-zone=Asia/Shanghai

#

spring.thymeleaf.cache=true

spring.thymeleaf.enabled=true

file.upload.path=D:/Project/JavaWeb/SpringBoot/04JPASpringBoot/fileUpLoad

spring.servlet.multipart.enabled=true

spring.servlet.multipart.file-size-threshold=0

spring.servlet.multipart.location=D:/Project/JavaWeb/SpringBoot/04JPASpringBoot/temp

spring.servlet.multipart.max-file-size=10MB

spring.servlet.multipart.max-request-size=10MB

spring.servlet.multipart.resolve-lazily=false

#

# debug=true # Enable debug logs.

# trace=true # Enable trace logs.

# LOGGING

logging.config=classpath:logback.xml

# zookeeper

zk.url=127.0.0.1:2181

zk.retry-time-init = 1000

zk.retry-count = 3

启动类

package com.smartmap.sample.test;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication

public class TestZookeeperApplication {

    public static void main(String[] args) {

        SpringApplication.run(TestZookeeperApplication.class, args);

    }

}

Zookeeper配置类

package com.smartmap.sample.test.conf;

import java.util.ArrayList;

import java.util.Collection;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.curator.RetryPolicy;

import org.apache.curator.framework.CuratorFramework;

import org.apache.curator.framework.CuratorFrameworkFactory;

import org.apache.curator.framework.api.CuratorEvent;

import org.apache.curator.framework.api.CuratorEventType;

import org.apache.curator.framework.api.CuratorListener;

import org.apache.curator.retry.ExponentialBackoffRetry;

import org.apache.curator.x.discovery.ServiceDiscovery;

import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;

import org.apache.curator.x.discovery.ServiceInstance;

import org.apache.curator.x.discovery.ServiceInstanceBuilder;

import org.apache.curator.x.discovery.details.JsonInstanceSerializer;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher.Event.EventType;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class ZookeeperConfiguration {

    Log log = LogFactory.getLog(ZookeeperConfiguration.class);

    @Value("${zk.url}")

    private String zkUrl;

    @Value("${zk.retry-time-init}")

    private int retryTimeInit;

    @Value("${zk.retry-count}")

    private int retryCount;

    @Bean

    public CuratorFramework getCuratorFramework() throws Exception {

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(retryTimeInit, retryCount);

        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zkUrl, retryPolicy);

        addListener(curatorFramework);

        curatorFramework.start();

        registerService(curatorFramework);

        ServiceInstance<ServerPayload> service = findService(curatorFramework, "book");

        log.info(service);

        return curatorFramework;

    }

    /**

     * 添加监听

     * @param curatorFramework

     */

    private void addListener(CuratorFramework curatorFramework) {

        curatorFramework.getCuratorListenable().addListener(new CuratorListener() {

            @Override

            public void eventReceived(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {

                CuratorEventType type = curatorEvent.getType();

                if(type==CuratorEventType.WATCHED) {

                    WatchedEvent we = curatorEvent.getWatchedEvent();

                    EventType et = we.getType();

                    if(we.getPath() != null) {

                        log.info(et + ":" + we.getPath());
                        // 重新注册监听

                        curatorFramework.checkExists().watched().forPath(we.getPath());

                    }

                }

            }

        });

    }

    /**

     * 服务注册

     * @param client

     * @throws Exception

     */

    protected void registerService(CuratorFramework client) throws Exception {

        // 构造一个服务描述

        ServiceInstanceBuilder<ServerPayload> serviceInstanceBuilder = ServiceInstance.builder();

        serviceInstanceBuilder.address("127.0.0.1");

        serviceInstanceBuilder.port(8080);

        serviceInstanceBuilder.name("book");

        ServerPayload serverPayload = new ServerPayload();

        serverPayload.setUrl("/api/v1.1/system/book");

        serviceInstanceBuilder.payload(serverPayload);

        ServiceInstance<ServerPayload> serviceInstance = serviceInstanceBuilder.build();

        // 服务注册

        /*JsonInstanceSerializer<ServerPayload> jsonInstanceSerializer = new JsonInstanceSerializer(ServerPayload.class);

        ServiceDiscovery<ServerPayload> serviceDiscovery = ServiceDiscoveryBuilder.builder(ServerPayload.class).client(client)

                .serializer(jsonInstanceSerializer).basePath("/service").build();

        serviceDiscovery.registerService(serviceInstance);

        serviceDiscovery.start();*/

        ServiceDiscovery<ServerPayload> serviceDiscovery = getServiceDiscovery(client);

        serviceDiscovery.registerService(serviceInstance);

    }

    /**

     * 服务发现

     * @param client

     * @param serviceName

     * @return

     * @throws Exception

     */

    protected ServiceInstance<ServerPayload> findService(CuratorFramework client, String serviceName) throws Exception {

        /*JsonInstanceSerializer<ServerPayload> jsonInstanceSerializer = new JsonInstanceSerializer(ServerPayload.class);

        ServiceDiscovery<ServerPayload> serviceDiscovery = ServiceDiscoveryBuilder.builder(ServerPayload.class).client(client)

                .serializer(jsonInstanceSerializer).basePath("/service").build();

        serviceDiscovery.start();*/

        ServiceDiscovery<ServerPayload> serviceDiscovery = getServiceDiscovery(client);

        // 查找服务

        Collection<ServiceInstance<ServerPayload>> all = serviceDiscovery.queryForInstances(serviceName);

        if(all.size() == 0) {

            return null;

        }

        else {

            // 取第一个服务

            ServiceInstance<ServerPayload> service = new ArrayList<ServiceInstance<ServerPayload>>(all).get(0);

            log.info(service.getAddress());

            log.info(service.getPayload());

            return service;

        }

    }    

    public ServiceDiscovery<ServerPayload> getServiceDiscovery(CuratorFramework client) throws Exception {

        JsonInstanceSerializer<ServerPayload> jsonInstanceSerializer = new JsonInstanceSerializer<ServerPayload>(ServerPayload.class);

        ServiceDiscovery<ServerPayload> serviceDiscovery = ServiceDiscoveryBuilder.builder(ServerPayload.class).client(client)

                .serializer(jsonInstanceSerializer).basePath("/service").build();

        serviceDiscovery.start();

        return serviceDiscovery;

    }

    public static class ServerPayload {

        private String url;

        public String getUrl() {

            return url;

        }

        public void setUrl(String url) {

            this.url = url;

        }        

    }

}

CuratorFrameWork调用Zookeeper

package com.smartmap.sample.test.service.impl;

import java.util.List;

import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.curator.framework.CuratorFramework;

import org.apache.curator.framework.recipes.locks.InterProcessMutex;

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.data.Stat;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import com.smartmap.sample.test.service.OrderService;

@Service

public class OrderServiceImpl implements OrderService {

    final Log log = LogFactory.getLog(OrderServiceImpl.class);

    final String lockPath = "/lock/order";

    @Autowired

    CuratorFramework curatorFramework;

    /**

     * 创建节点

     */

    public String createPath(String path) throws Exception {

        curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath(path, new byte[0]);

        return "create " + path;

    }

    /**

     * 删除节点

     */

    public String delete(String path) throws Exception {

        curatorFramework.delete().forPath(path);

        return "delete " + path;

    }

    /**

     * 获取节点数据

     */

    public String getData(String path) throws Exception {

        byte[] bs = curatorFramework.getData().forPath(path);

        String result = new String(bs);

        return "get data: " + result;

    }

    /**

     * 设置节点数据

     */

    public String setData(String path, String data) throws Exception {

        curatorFramework.setData().forPath(path, data.getBytes());

        return "set data " + data;

    }

    /**

     * 检查节点是否存在

     */

    public String check(String path) throws Exception {

        Stat stat = curatorFramework.checkExists().forPath(path);

        return "stat " + stat;

    }

    /**

     * 获取子节点

     */

    public String children(String path) throws Exception {

        List<String> children = curatorFramework.getChildren().forPath(path);

        return "children " + children;

    }

    /**

     * 监控节点

     */

    public String watch(String path) throws Exception {

        Stat stat = curatorFramework.checkExists().watched().forPath(path);

        return "watch： < " + path + " >    stat: " + stat;

    }

    /**

     * 分布式锁

     */

    public void makeOrderType(String type) {

        String path = lockPath + "/" + type;

        log.info("try do job for " + type);

        try {

            InterProcessMutex lock = new InterProcessMutex(curatorFramework, path);

            if (lock.acquire(10, TimeUnit.HOURS)) {

                try {

                    Thread.sleep(1000 * 5);

                    log.info("do job " + type + " done");

                } finally {

                    lock.release();

                }

            }

        } catch (Exception ex) {

            ex.printStackTrace();

        }

    }

}

Controller类

package com.smartmap.sample.test.controller.rest;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.PathVariable;

import org.springframework.web.bind.annotation.PostMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.RestController;

import com.smartmap.sample.test.service.OrderService;

@RestController

@RequestMapping("/api/v1.1/system/book")

public class ZookeeperTestCrontroller {

    @Autowired

    OrderService orderService;

    /**

     * 创建节点

     *

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/create?path=/testNode'

     * @param path

     * @return

     * @throws Exception

     */

    @PostMapping("/create")

    public String create(@RequestParam("path") String path) throws Exception {

        return orderService.createPath(path);

    }

    /**

     * 删除节点

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/delete?path=/testNode'

     * @param path

     * @return

     * @throws Exception

     */

    @PostMapping("/delete")

    public String delete(@RequestParam("path") String path) throws Exception {

        return orderService.delete(path);

    }

    /**

     * 获取节点数据

     * curl -XGET 'http://127.0.0.1:8080/test/api/v1.1/system/book/data?path=/testNode'

     * @param path

     * @return

     * @throws Exception

     */

    @GetMapping("/data")

    public String getData(@RequestParam("path") String path) throws Exception {

        return orderService.getData(path);

    }

    /**

     * 设置节点数据

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/data?path=/testNode&data=987654321'

     * @param path

     * @param data

     * @return

     * @throws Exception

     */

    @PostMapping("/data")

    public String setData(@RequestParam("path") String path, @RequestParam("data") String data) throws Exception {

        return orderService.setData(path, data);

    }

    /**

     * 检查节点是否存在

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/check?path=/testNode'

     * @param path

     * @return

     * @throws Exception

     */

    @PostMapping("/check")

    public String check(@RequestParam("path") String path) throws Exception {

        return orderService.check(path);

    }

    /**

     * 获取子节点

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/children?path=/testNode'

     * @param path

     * @return

     * @throws Exception

     */

    @PostMapping("/children")

    public String getChildren(@RequestParam("path") String path) throws Exception {

        return orderService.children(path);

    }

    /**

     * 监控节点

     *

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/watch?path=/testNode'

     * @param path

     * @return

     * @throws Exception

     */

    @PostMapping("/watch")

    public String watch(@RequestParam("path") String path) throws Exception {

        return orderService.watch(path);

    }

    /**

     * 订单--分布式锁

     * curl -XPOST 'http://127.0.0.1:8080/test/api/v1.1/system/book/order?type=orderType'

     * @param type

     * @return

     * @throws Exception

     */

    @PostMapping("/order")

    public String makeOrder(@RequestParam("type") String type) throws Exception {

         orderService.makeOrderType(type);

         return "success";

    }

}

巴特西

Spring Boot—20Zookeeper

最新文章

热门文章