分布式缓存技术Redis的实战

//源码

private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {
HostAndPort master = null;
boolean sentinelAvailable = false;
log.info("Trying to find master from available Sentinels...");
// 有多个sentinels,遍历这些个sentinels
for (String sentinel : sentinels) {
// host:port表示的sentinel地址转化为一个HostAndPort对象。
final HostAndPort hap = HostAndPort.parseString(sentinel);
log.fine("Connecting to Sentinel " + hap);
Jedis jedis = null;
try {
// 连接到sentinel
jedis = new Jedis(hap.getHost(), hap.getPort());
// 根据masterName得到master的地址，返回一个list，host= list[0], port =// list[1]
List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
// connected to sentinel...
sentinelAvailable = true;if (masterAddr == null || masterAddr.size() != 2) {
log.warning("Can not get master addr, master name: " + masterName + ".
Sentinel: " + hap
+ ".");
continue;
}
// 如果在任何一个sentinel中找到了master，不再遍历sentinels
master = toHostAndPort(masterAddr);
log.fine("Found Redis master at " + master);
break;
} catch (JedisException e) {
// resolves #1036, it should handle JedisException there's another chance
// of raising JedisDataException
log.warning("Cannot get master address from sentinel running @ " + hap + ".
Reason: " + e
+ ". Trying next one.");
} finally {
if (jedis != null) {
jedis.close();
}
}
}
// 到这里，如果master为null，则说明有两种情况，一种是所有的sentinels节点都down掉了，一种是master节点没有被存活的sentinels监控到
if (master == null) {
if (sentinelAvailable) {
// can connect to sentinel, but master name seems to not
// monitored
throw new JedisException("Can connect to sentinel, but " + masterName
+ " seems to be not monitored...");
} else {
throw new JedisConnectionException("All sentinels down, cannot determine where is
"
+ masterName + " master is running...");
}
}
//如果走到这里，说明找到了master的地址
log.info("Redis master running at " + master + ", starting Sentinel listeners...");
//启动对每个sentinels的监听为每个sentinel都启动了一个监听者MasterListener。MasterListener本身是一个线程，它会去订阅sentinel上关于master节点地址改变的消息。
for (String sentinel : sentinels) {
final HostAndPort hap = HostAndPort.parseString(sentinel);
MasterListener masterListener = new MasterListener(masterName, hap.getHost(),
hap.getPort());
// whether MasterListener threads are alive or not, process can be stopped
masterListener.setDaemon(true);
masterListeners.add(masterListener);
masterListener.start();
} r
eturn master;
}

public List<String> sentinelGetMasterAddrByName(String masterName) {
client.sentinel(Protocol.SENTINEL_GET_MASTER_ADDR_BY_NAME, masterName);
final List<Object> reply = client.getObjectMultiBulkReply();
return BuilderFactory.STRING_LIST.build(reply);
}

//链接方式
Set<HostAndPort> hostAndPorts=new HashSet<>();
HostAndPort hostAndPort=new HostAndPort("192.168.11.153",7000);
HostAndPort hostAndPort1=new HostAndPort("192.168.11.153",7001);
HostAndPort hostAndPort2=new HostAndPort("192.168.11.154",7003);
HostAndPort hostAndPort3=new HostAndPort("192.168.11.157",7006);
hostAndPorts.add(hostAndPort);
hostAndPorts.add(hostAndPort1);
hostAndPorts.add(hostAndPort2);
hostAndPorts.add(hostAndPort3);
JedisCluster jedisCluster=new JedisCluster(hostAndPorts,6000);
jedisCluster.set("mic","hello");

public List<String> sentinelGetMasterAddrByName(String masterName) {
client.sentinel(Protocol.SENTINEL_GET_MASTER_ADDR_BY_NAME, masterName);
final List<Object> reply = client.getObjectMultiBulkReply();
return BuilderFactory.STRING_LIST.build(reply);
} S
et<HostAndPort> hostAndPorts=new HashSet<>();
HostAndPort hostAndPort=new HostAndPort("192.168.11.153",7000);
HostAndPort hostAndPort1=new HostAndPort("192.168.11.153",7001);
HostAndPort hostAndPort2=new HostAndPort("192.168.11.154",7003);
HostAndPort hostAndPort3=new HostAndPort("192.168.11.157",7006);
hostAndPorts.add(hostAndPort);
hostAndPorts.add(hostAndPort1);
hostAndPorts.add(hostAndPort2);
hostAndPorts.add(hostAndPort3);
JedisCluster jedisCluster=new JedisCluster(hostAndPorts,6000);
jedisCluster.set("mic","hello");

Config config=new Config();
config.useClusterServers().setScanInterval(2000).addNodeAddress("redis://192.168.11.153:7000",
"redis://192.168.11.153:7001",
"redis://192.168.11.154:7003","redis://192.168.11.157:7006");

RedissonClient redissonClient= Redisson.create(config);

RBucket<String> rBucket=redissonClient.getBucket("mic");

System.out.println(rBucket.get());

public class UnitTest extends Thread{

    @Override
    public void run() {
        while(true){
            DistributedLock distributedLock=new DistributedLock();
            String rs=distributedLock.acquireLock("updateOrder",
                    2000,5000);
            if(rs!=null){
                System.out.println(Thread.currentThread().getName()+"-> 成功获得锁:"+rs);
                try {
                    Thread.sleep(1000);
                    distributedLock.releaseLockWithLua("updateOrder",rs);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                break;
            }
        }
    }

    public static void main(String[] args) {
        UnitTest unitTest=new UnitTest();
        for(int i=0;i<10;i++){
            new Thread(unitTest,"tName:"+i).start();
        }
    }
}

public class DistributedLock {

    //获得锁

    /**
     *
     * @param lockName  锁的名词
     * @param acquireTimeout  获得锁的超时时间
     * @param lockTimeout 锁本身的过期时间
     * @return
     */
    public String acquireLock(String lockName,long acquireTimeout,long lockTimeout){
        String identifier=UUID.randomUUID().toString();//保证释放锁的时候是同一个持有锁的人
        String lockKey="lock:"+lockName;
        int lockExpire=(int)(lockTimeout/1000);
        Jedis jedis=null;
        try {
            jedis = JedisConnectionUtils.getJedis();
            long end = System.currentTimeMillis() + acquireTimeout;
            //获取锁的限定时间
            while (System.currentTimeMillis() < end) {
                if (jedis.setnx(lockKey, identifier) == 1) { //设置值成功
                    jedis.expire(lockKey, lockExpire); //设置超时时间
                    return identifier; //获得锁成功
                }
                //
                if (jedis.ttl(lockKey) == -1) {
                    jedis.expire(lockKey, lockExpire); //设置超时时间
                }
                try {
                    //等待片刻后进行获取锁的重试
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }finally {
            jedis.close(); //回收
        }
        return null;
    }

    public boolean releaseLockWithLua(String lockName,String identifier){
        System.out.println(lockName+"开始释放锁："+identifier);

        Jedis jedis=JedisConnectionUtils.getJedis();
        String lockKey="lock:"+lockName;

        String lua="if redis.call(\"get\",KEYS[1])==ARGV[1] then " +
                "return redis.call(\"del\",KEYS[1]) " +
                "else return 0 end";
        Long rs=(Long) jedis.eval(lua,1,new String[]{lockKey,identifier});
        if(rs.intValue()>0){
            return true;
        }
        return false;

    }

    //释放锁
    public boolean releaseLock(String lockName,String identifier){
        System.out.println(lockName+"开始释放锁："+identifier);

        String lockKey="lock:"+lockName;
        Jedis jedis=null;
        boolean isRelease=false;
        try{
            jedis=JedisConnectionUtils.getJedis();
            while(true){
                jedis.watch(lockKey);
                //判断是否为同一把锁
                if(identifier.equals(jedis.get(lockKey))){
                    Transaction transaction=jedis.multi();
                    transaction.del(lockKey);
                    if(transaction.exec().isEmpty()){
                        continue;
                    }
                    isRelease=true;
                }
                //TODO 异常
                jedis.unwatch();
                break;
            }
        }finally {
            jedis.close();
        }
        return  isRelease;
    }

}

public class JedisConnectionUtils {
    private static JedisPool pool=null;
    static {
        JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(100);
        pool=new JedisPool(jedisPoolConfig,"192.168.11.153",6379);
    }
    public static Jedis getJedis(){
        return pool.getResource();
    }
}

public class RedissonLock {

    public static void main(String[] args) {
        Config config=new Config();
        config.useSingleServer().setAddress("redis://192.168.11.153:6379");
        RedissonClient redissonClient=Redisson.create(config);
        RLock rLock=redissonClient.getLock("updateOrder");
        try {
            rLock.tryLock(100,10,TimeUnit.SECONDS);
            System.out.println("test");
            Thread.sleep(1000);
            rLock.unlock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            rLock.unlock();
            redissonClient.shutdown();
        }
    }
}

Jedis jedis=new Jedis("192.168.11.152",6379);
Pipeline pipeline=jedis.pipelined();
for(int i=0;i<1000;i++){
pipeline.incr("test");
}
pipeline.sync();