Redis项目实战（三）---Jedis使用

Jedis即redis java客户端，源码地址：https://github.com/xetorthio/jedis

pom配置：

<dependency>

   <groupId>redis.clients</groupId>

    <artifactId>jedis</artifactId>

    <version>3.0.1</version>

    <type>jar</type>

    <scope>compile</scope>

</dependency>

 

 

1）链接单机版

     public void testJedis() throws Exception {
          // 第一步：建立一个Jedis对象。须要指定服务端的ip及端口。
          Jedis jedis = new Jedis("192.168.**.**", 6379);
          // 第二步：使用Jedis对象操做数据库，每一个redis命令对应一个方法。
          String result = jedis.get("hello");
          // 第三步：打印结果。
          System.out.println(result);
          // 第四步：关闭Jedis
          jedis.close();
     }

2） 链接单机版使用链接池

     public void testJedisPool() throws Exception {
          // 第一步：建立一个JedisPool对象。须要指定服务端的ip及端口。
          JedisPool jedisPool = new JedisPool("192.168.**.**", 6379);
          // 第二步：从JedisPool中得到Jedis对象。
          Jedis jedis = jedisPool.getResource();
          // 第三步：使用Jedis操做redis服务器。
          jedis.set("jedis", "test");
          String result = jedis.get("jedis");
          System.out.println(result);
          // 第四步：操做完毕后关闭jedis对象，链接池回收资源。
          jedis.close();
          // 第五步：关闭JedisPool对象。
          jedisPool.close();
     }

3） 链接集群版

     public void testJedisCluster() throws Exception {
          // 第一步：使用JedisCluster对象。须要一个Set<HostAndPort>参数。Redis节点的列表。
          Set<HostAndPort> nodes = new HashSet<>();
          nodes.add(new HostAndPort("192.168.**.**", 7001));
          nodes.add(new HostAndPort("192.168.**.**", 7002));
          nodes.add(new HostAndPort("192.168.**.**", 7003));
          nodes.add(new HostAndPort("192.168.**.**", 7004));
          nodes.add(new HostAndPort("192.168.**.**", 7005));
          nodes.add(new HostAndPort("192.168.**.**", 7006));
          JedisCluster jedisCluster = new JedisCluster(nodes);
          // 第二步：直接使用JedisCluster对象操做redis。在系统中单例存在。
          jedisCluster.set("hello", "100");
          String result = jedisCluster.get("hello");
          // 第三步：打印结果
          System.out.println(result);
          // 第四步：系统关闭前，关闭JedisCluster对象。
          jedisCluster.close();
     }

 

4）JedisPool初始化配置：

private static final int taskCount = 50; // 并发任务 
private static final int batchSize = 10; // pipeline大小 
private static final int cmdCount = 1000;// 每一个任务处理命令数 
private static final boolean usePipeline = true;
 
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxActive(200);
poolConfig.setMaxIdle(100);
poolConfig.setMaxWait(2000);
poolConfig.setTestOnBorrow(false);
poolConfig.setTestOnReturn(false);
jedisPool = new JedisPool(poolConfig, host, port);

5）Jedis 消息发布与订阅API

   5.1）消息订阅

　　　

   public void subscribeChannal(){
       //从jedisPool中获取一个jedis对象
       Jedis jds = RedisInsUtil.getJedis();
       // 方式1 订阅获得信息在lister的onPMessage(...)方法中进行处理  
       jds.psubscribe(this, new String[]{"channal01","channal02"});

　　    // 方式2 订阅获得信息在lister的onMessage(...)方法中进行处理  (普通订阅方式)
　　    //jedis.subscribe(listener, "foo", "watson");  
    }
     
    // 初始化订阅时候的处理 
    @Override
    public void onSubscribe(String channel, int subscribedChannels) { 
    } 
     
     // 初始化按表达式的方式订阅时候的处理
     @Override
     public void onPSubscribe(String pattern, int subscribedChannels) {
     }

     @Override
     // 取得按表达式的方式订阅的消息后的处理 
     public void onPMessage(String pattern, String channel, String message) {
         LOG.info("onPMessage()," + pattern + "=" + channel + ",msg="+ message);
     }
     
     // 取得订阅的消息后的处理 
     public void onMessage(String channel, String message) {
     }

     5.2）消息发布

             发布消息只用调用Jedis的publish(...)方法便可。

    Jedis jedis = ru.getConnection(); //获取一个jedis对象，自行封装工具类 
    jedis.publish("hello_foo", "bar123");  

 

6) 管道

原理：

    Redis客户端与Redis服务器之间使用TCP协议进行链接，管道（pipeline）能够一次性发送多条命令并在执行完后一次性将结果返回，pipeline经过减小客户端与redis的通讯次数来实现下降往返延时时间，并且Pipeline 实现的原理是队列，而队列的原理是时先进先出，这样就保证数据的顺序性。 

    缺陷：须要注意到是用 pipeline方式打包命令发送，redis必须在处理完全部命令前先缓存起全部命令的处理结果。打包的命令越多，缓存消耗内存也越多。因此并非打包的命令越多越好。具体多少合适须要根据具体状况测试。

   备注：Pipeline 的默认的同步的个数为53个，也就是说arges中累加到53条数据时会把数据提交；

  

7）扩展：

    因为redis-cluster的hash分片，JedisCluster对象原生并不支持Pipline管道和keys方法。然而，实际项目中在追求性能以及考虑到Redis集群版和单机版兼容或切换问题，每每须要支持这些方法，经过整理网上的部分资料，现给出如下思路，具体方法可参考gitHub示例

   

  管道Pipline实现思路：

 

  Redis集群规范: Redis 集群的键空间被分割为  16384 个槽（slot）， 集群的最大节点数量也是  16384 个。每一个主节点都负责处理  16384 个哈希槽的其中一部分。当一个集群处于“稳定”（stable）状态时（群没有在执行重配置（reconfiguration）操做）， 每一个哈希槽都只由一个节点进行处理(hash Slot的分配是由CRC16算法计算)。

 故， 

  1.根据要插入的key知道这个key所对应的槽的号码（JedisClusterCRC16.getSlot(key)）, 再经过这个槽的号码从集群中找到对应Jedis（经过每一个节点的slot分布，就知道了哪些key应该在哪些节点上）。

  2.相同槽位的key，使用同一个jedis.pipeline去执行命令。

  3.合并这次pipeline全部的response返回。

 

  keys实现思路：

     循环集群中全部的节点（分别获取j对应的client对象），而后每一个节点作keys，最后再加到一块返回。

   

  备注：每次执行前须要刷新以获取最新的slot分布。