一次 HashSet 所引起的并发问题

//线程池
private ExecutorService executor;

private Set<String> set = new hashSet();

private void execute(){

    while(true){
        //从 MQ 中获取数据
        String key = subMQ();
        executor.excute(new Worker(key)) ;
    }
}

public class Worker extends Thread{
    private String key ;

    public Worker(String key){
        this.key = key;
    }

    @Override
    private void run(){
        if(!set.contains(key)){

            //数据库查询
            if(queryDB(key)){
                set.add(key);
                return;
            }
        }

        //达到某种条件时清空 set
        if(flag){
            set = null ;
        }
    }   
}