秒发4000万数据的缓存方案以及消息优先消费

  /**
   * 判断字符串是否在单片位数组中存在
   * @param set 单片位数组
   * @param mod 某个hash函数的值
   * @param hash 字符串转化后的hashcode
   */
  public static boolean setBitMap(int[] set, long mod, long hash){
    boolean isExist = false;
    /**字符串在位数组中的位置*/
    long bit = hash%mod;
    /**int类型的位置*/
    long site = bit >> 5;
    /**int类型中32位的第mod1位*/
    long mod1 = bit & 31;
    /**位数组中该字符串的位置是否为1
     * 如果是则退出时exits+1，否则将该位置变成1
     * */
    try {
      if(((set[(int) site] >> mod1) & 1) == 1){
        isExist = true;
      }
      else {
        set[(int) site] = set[(int) site] | (1 << mod1);
      }  
    } catch (Exception e) {
    }
    return isExist;
  }

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.support.collections.DefaultRedisList;
import org.springframework.stereotype.Component;
import java.io.Serializable;
import java.util.StringJoiner;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

@Component
public class RedisQueue implements ApplicationListener<ContextRefreshedEvent> {

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
//        addTestData();
        new Thread(() -> {
        //优化成线程池方式异步执行，千万不要阻塞主线程
            while (true){
                try {
                    handler((message -> {
                        System.out.println(message);
                        return null;
                    }),(message -> {
                        try {
                        //理论是永远不会执行
                            System.out.println("no message wait");
                            Thread.sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return null;
                    }),maxAll,max,middle,min);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
    public void handler(Function<Message,Object> callBack,Function<Message,Object> noMessageCallBack,DefaultRedisList<Message> ...queues) throws Exception {
        notify.take();
        for(DefaultRedisList<Message> queue : queues){
            if(queue.size() > 0){
                Message take = queue.take();
                callBack.apply(take);
                return;
            }
        }
        noMessageCallBack.apply(null);
    }

    private void addTestData() {
        for(int i = 0 ;i < 2 ; i ++){
            maxAll.add(new Message("maxAll:" + UUID.randomUUID().toString(),"maxAll"));
        }

        for(int i = 0 ;i < 100 ; i ++){
            max.add(new Message("max:" + UUID.randomUUID().toString(),"max"));
        }

        for(int i = 0 ;i < 1000 ; i ++){
            middle.add(new Message("middle:" + UUID.randomUUID().toString(),"middle"));
        }

        for(int i = 0 ;i < 10000 ; i ++){
            min.add(new Message("min:" + UUID.randomUUID().toString(),"min"));
        }
    }

    static class Message implements Serializable {
        private String msg;
        private String type;

        @Override
        public String toString() {
            return new StringJoiner(", ", Message.class.getSimpleName() + "[", "]")
                    .add("msg='" + msg + "'")
                    .add("type='" + type + "'")
                    .toString();
        }

        public Message(){
        }

        public Message(String msg, String type) {
            this.msg = msg;
            this.type = type;
        }

        public String getMsg() {
            return msg;
        }

        public void setMsg(String msg) {
            this.msg = msg;
        }

        public String getType() {
            return type;
        }

        public void setType(String type) {
            this.type = type;
        }
    }

    public static final String PRIORIT_MAX_ALL = "MAX_ALL";
    public static final String PRIORIT_MAX = "MAX";
    public static final String PRIORIT_MIDDLE = "MIDDLE";
    public static final String PRIORIT_MIN = "MIN";

    public static final String NOTIFY = "notify";

    private NotifyDefaultRedisList<Message> maxAll,max,min,middle;
    private DefaultRedisList<String> notify;

    public RedisQueue(@Autowired ApplicationContext context, @Autowired RedisTemplate redisTemplate){
        notify = new DefaultRedisList(NOTIFY,redisTemplate);
        maxAll = new NotifyDefaultRedisList(notify,PRIORIT_MAX_ALL,redisTemplate);
        max = new NotifyDefaultRedisList(notify,PRIORIT_MAX,redisTemplate);
        middle = new NotifyDefaultRedisList(notify,PRIORIT_MIDDLE,redisTemplate);
        min = new NotifyDefaultRedisList(notify,PRIORIT_MIN,redisTemplate);
    }

    static class NotifyDefaultRedisList<T> extends DefaultRedisList<T> {

        private DefaultRedisList<String> notify;

        public NotifyDefaultRedisList(DefaultRedisList<String> notify,String key, RedisOperations<String, T> operations) {
            super(key, operations);
            this.notify = notify;
        }

        @Override
        public boolean add(T value) {
            this.notify.add(UUID.randomUUID().toString());
            return super.add(value);
        }

    }
}

https://www.oschina.net/question/8729_224337