Merge pull request #2734 from gabrywu/removeUnnecessaryLock

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

import static org.apache.dolphinscheduler.common.Constants.COMMA;

     */

    private ConcurrentHashMap<String, Set<HostWeight>> workerHostWeights;

    /**

     * worker group host lock

     */

    private Lock lock;

    /**

     * executor service

     */

    public void init(){

        this.selector = new LowerWeightRoundRobin();

        this.workerHostWeights = new ConcurrentHashMap<>();

        this.lock = new ReentrantLock();

        this.executorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LowerWeightHostManagerExecutor"));

        this.executorService.scheduleWithFixedDelay(new RefreshResourceTask(),35, 40, TimeUnit.SECONDS);

        this.roundRobinHostManager = new RoundRobinHostManager();

    }

    private void syncWorkerHostWeight(Map<String, Set<HostWeight>> workerHostWeights) {

        lock.lock();

        try {

            workerHostWeights.clear();

            workerHostWeights.putAll(workerHostWeights);

        } finally {

            lock.unlock();

        }

        this.workerHostWeights.clear();

        this.workerHostWeights.putAll(workerHostWeights);

    }

    private Set<HostWeight> getWorkerHostWeights(String workerGroup) {

        lock.lock();

        try {

        return workerHostWeights.get(workerGroup);

        } finally {

            lock.unlock();

        }

    }

    class RefreshResourceTask implements Runnable{