美文网首页Redisson
实操Redission 分布式对象(一)

实操Redission 分布式对象(一)

作者: 周二倩你一生 | 来源:发表于2020-04-02 14:37 被阅读0次

    公共讲解

    RedissonClient(同步异步)、RedissonReactiveClient(反射式Reactive
    )和RedissonRxClient(RxJava2)实例本身和Redisson提供的所有分布式对象都是线程安全的

    1.首先先进行一步简单的配置redis的步骤

    Config config = new Config();
    SingleServerConfig singleServerConfig = config.useSingleServer()
            .setAddress("redis://" + StringUtils.trim("10.6.30.41:6379"))
            .setDatabase(8)
            .setTimeout(3000)
            .setConnectionPoolSize(64)
            .setConnectionMinimumIdleSize(10);
        singleServerConfig.setPassword("123456");
    RedissonClient client = Redisson.create(config);
    

    2.了解常用RKeys的API操作

    RKeys keys = client.getKeys();
    //获取所有key值
    Iterable<String> allKeys = keys.getKeys();
    //获取所有模糊key值
    Iterable<String> foundedKeys = keys.getKeysByPattern("key");
    //删除多个key值
    long numOfDeletedKeys = keys.delete("obj1", "obj2", "obj3");
    //删除模糊key值
    long deletedKeysAmount = keys.deleteByPattern("test?");
    //随机获取key
    String randomKey = keys.randomKey();
    //当前多少key数
    long keysAmount = keys.count();
    

    3.通用对象桶(Object Bucket)

    Redisson的分布式RBucketJava对象是一种通用对象桶可以用来存放任类型的对象。 除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    //对象桶 单个元素的存储
    RBucket<String> bucket = client.getBucket("REDIS_BUCKET_TEST");
    //同步 存储元素进对象桶,无返回值
    bucket.set("1");
    //异步
    bucket.setAsync("1");
    //返回bucket中key的值,返回1
    bucket.get();
    //尝试对REDIS_BUCKET_TEST存值,如果已存在值就不存进去,存在就存进去
    // (先查询是否存在,true则存入,false则返回)
    boolean bl = bucket.trySet("2");
    //第一个参数与存的值比较,相等就存入第二个参数
    boolean bl = bucket.compareAndSet("2", "3");
    //get返回并且存入新的值
    String str =  bucket.getAndSet("4");
    

    还可以通过RBuckets接口实现批量操作多个RBucket对象:

    RBuckets buckets = client.getBuckets();
    //获取多个buckets
    Map<String, String> loadedBuckets = buckets.get("REDIS_BUCKET_TEST", "REDIS_BUCKET_TEST1", "REDIS_BUCKET_TEST2");
    Map<String, Object> map = new HashMap<>();
    map.put("REDIS_BUCKET_TEST", "0");
    map.put("REDIS_BUCKET_TEST1","1");
    // 利用Redis的事务特性,同时保存所有的通用对象桶,如果任意一个通用对象桶已经存在则放弃保存其他所有数据。
    buckets.trySet(map);
    // 同时保存全部通用对象桶。
    buckets.set(map);
    

    4.二进制流(Binary Stream)

    Redisson的分布式RBinaryStream Java对象同时提供了InputStream接口和OutputStream接口的实现。流的最大容量受Redis主节点的内存大小限制。

    RBinaryStream stream = client.getBinaryStream("anyStream");
    byte[] content = "ceshi".getBytes();
    //设置流内容ceshi
    stream.set(content);
    InputStream is = stream.getInputStream();
    byte[] readBuffer = new byte[512];
    //读取流ceshi
    is.read(readBuffer);
    OutputStream os = stream.getOutputStream();
    byte[] contentToWrite = "ceshi1".getBytes();
    //写入流ceshi1,注意:不是覆盖是写入,最终流为ceshiceshi1
    os.write(contentToWrite);
    

    5.地理空间对象桶(Geospatial Bucket)

    RGeo<String> geo = client.getGeo("test");
    //同步存储可存储多个
    geo.add(new GeoEntry(13.361389, 38.115556, "Palermo"),
            new GeoEntry(15.087269, 37.502669, "Catania"));
    //异步存储只能一个
    geo.addAsync(37.618423, 55.751244, "Moscow");
    //返回两个坐标的距离
    Double distance = geo.dist("Palermo", "Catania", GeoUnit.METERS);
    //返回成员的地理位置
    Map<String, GeoPosition> positions = geo.pos("test2", "Palermo", "test3", "Catania", "test1");
    //暂时没研究
    RFuture<Map<String, String>> future= geo.hashAsync("Palermo", "Catania");
    List<String> cities = geo.radius(15, 37, 200, GeoUnit.KILOMETERS);
    Map<String, GeoPosition> citiesWithPositions = geo.radiusWithPosition(15, 37, 200, GeoUnit.KILOMETERS);
    

    6.BitSet(暂时还不理解)

    Redisson的分布式RBitSetJava对象采用了与java.util.BiteSet类似结构的设计风格。可以理解为它是一个分布式的可伸缩式位向量。需要注意的是RBitSet的大小受Redis限制,最大长度为4 294 967 295。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    RBitSet set = redisson.getBitSet("simpleBitset");
    set.set(0, true);
    set.set(1812, false);
    set.clear(0);
    set.addAsync("e");
    set.xor("anotherBitset");
    

    BitSet数据分片(Sharding)(分布式RoaringBitMap)

    基于Redis的Redisson集群分布式BitSet通过RClusteredBitSet接口,为集群状态下的Redis环境提供了BitSet数据分片的功能。通过优化后更加有效的分布式RoaringBitMap算法,突破了原有的BitSet大小限制,达到了集群物理内存容量大小。在这里可以获取更多的内部信息。

    RClusteredBitSet set = redisson.getClusteredBitSet("simpleBitset");
    set.set(0, true);
    set.set(1812, false);
    set.clear(0);
    set.addAsync("e");
    set.xor("anotherBitset");
    

    7.原子整长形(AtomicLong)

    Redisson的分布式整长形RAtomicLong对象和Java中的java.util.concurrent.atomic.AtomicLong对象类似。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。
    个人理解:保证齐原子性,用了CAS算法乐观锁技术,

    RAtomicLong atomicLong = client.getAtomicLong("myAtomicLong");
    //设置成3
    atomicLong.set(3);
    //原子地将当前值增加1。
    atomicLong.incrementAndGet();
    //获取值
    atomicLong.get();
    

    8.原子双精度浮点(AtomicDouble)

    Redisson还提供了分布式原子双精度浮点RAtomicDouble,弥补了Java自身的不足。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    RAtomicDouble atomicDouble = redisson.getAtomicDouble("myAtomicDouble");
    //设置值
    atomicDouble.set(2.81);
    //加法并且返回加好后的值
    atomicDouble.addAndGet(4.11);
    //获取值
    atomicDouble.get();
    

    9.话题(订阅分发)

    Redisson的分布式话题RTopic对象实现了发布、订阅的机制。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    RTopic topic = redisson.getTopic("anyTopic");
    topic.addListener(SomeObject.class, new MessageListener<SomeObject>() {
        @Override
        public void onMessage(String channel, SomeObject message) {
            //...
        }
    });
    
    // 在其他线程或JVM节点
    RTopic topic = redisson.getTopic("anyTopic");
    long clientsReceivedMessage = topic.publish(new SomeObject());
    

    在Redis节点故障转移(主从切换)或断线重连以后,所有的话题监听器将自动完成话题的重新订阅。

    模糊话题

    Redisson的模糊话题RPatternTopic对象可以通过正式表达式来订阅多个话题。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    // 订阅所有满足`topic1.*`表达式的话题
    RPatternTopic topic1 = redisson.getPatternTopic("topic1.*");
    int listenerId = topic1.addListener(Message.class, new PatternMessageListener<Message>() {
        @Override
        public void onMessage(String pattern, String channel, Message msg) {
             Assert.fail();
        }
    });
    

    在Redis节点故障转移(主从切换)或断线重连以后,所有的模糊话题监听器将自动完成话题的重新订阅。

    10.布隆过滤器(Bloom Filter)

    Redisson利用Redis实现了Java分布式布隆过滤器(Bloom Filter)。所含最大比特数量为2^32

    RBloomFilter<SomeObject> bloomFilter = redisson.getBloomFilter("sample");
    // 初始化布隆过滤器,预计统计元素数量为55000000,期望误差率为0.03
    bloomFilter.tryInit(55000000L, 0.03);
    bloomFilter.add(new SomeObject("field1Value", "field2Value"));
    bloomFilter.add(new SomeObject("field5Value", "field8Value"));
    bloomFilter.contains(new SomeObject("field1Value", "field8Value"));
    

    6.8.1. 布隆过滤器数据分片(Sharding)

    基于Redis的Redisson集群分布式布隆过滤器通过RClusteredBloomFilter接口,为集群状态下的Redis环境提供了布隆过滤器数据分片的功能。 通过优化后更加有效的算法,通过压缩未使用的比特位来释放集群内存空间。每个对象的状态都将被分布在整个集群中。所含最大比特数量为2^64。在这里可以获取更多的内部信息。

    RClusteredBloomFilter<SomeObject> bloomFilter = redisson.getClusteredBloomFilter("sample");
    // 采用以下参数创建布隆过滤器
    // expectedInsertions = 255000000
    // falseProbability = 0.03
    bloomFilter.tryInit(255000000L, 0.03);
    bloomFilter.add(new SomeObject("field1Value", "field2Value"));
    bloomFilter.add(new SomeObject("field5Value", "field8Value"));
    bloomFilter.contains(new SomeObject("field1Value", "field8Value"));
    

    该功能仅限于Redisson PRO版本。

    11.基数估计算法(HyperLogLog)

    实用场景:概率数据结构使您可以以极高的空间效率维护数百万个项目的计数。
    Redisson利用Redis实现了Java分布式基数估计算法(HyperLogLog)对象。该对象可以在有限的空间内通过概率算法统计大量的数据。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    RHyperLogLog<Integer> log = redisson.getHyperLogLog("log");
    log.add(1);
    log.add(2);
    log.add(3);
    
    log.count();
    

    12. 整长型累加器(LongAdder)

    基于Redis的Redisson分布式整长型累加器(LongAdder)采用了与java.util.concurrent.atomic.LongAdder类似的接口。通过利用客户端内置的LongAdder对象,为分布式环境下递增和递减操作提供了很高得性能。据统计其性能最高比分布式AtomicLong对象快 12000 倍。完美适用于分布式统计计量场景。

    RLongAdder atomicLong = redisson.getLongAdder("myLongAdder");
    atomicLong.add(12);
    atomicLong.increment();
    atomicLong.decrement();
    atomicLong.sum();
    

    当不再使用整长型累加器对象的时候应该自行手动销毁,如果Redisson对象被关闭(shutdown)了,则不用手动销毁。

    RLongAdder atomicLong = ...
    atomicLong.destroy();
    

    13.双精度浮点累加器(DoubleAdder)

    基于Redis的Redisson分布式双精度浮点累加器(DoubleAdder)采用了与java.util.concurrent.atomic.DoubleAdder类似的接口。通过利用客户端内置的DoubleAdder对象,为分布式环境下递增和递减操作提供了很高得性能。据统计其性能最高比分布式AtomicDouble对象快 12000 倍。完美适用于分布式统计计量场景。

    RLongDouble atomicDouble = redisson.getLongDouble("myLongDouble");
    atomicDouble.add(12);
    atomicDouble.increment();
    atomicDouble.decrement();
    atomicDouble.sum();
    

    当不再使用双精度浮点累加器对象的时候应该自行手动销毁,如果Redisson对象被关闭(shutdown)了,则不用手动销毁。

    RLongDouble atomicDouble = ...
    
    _b6d2063_
    atomicDouble.destroy();
    

    13.限流器(RateLimiter)

    基于Redis的分布式限流器(RateLimiter)可以用来在分布式环境下现在请求方的调用频率。既适用于不同Redisson实例下的多线程限流,也适用于相同Redisson实例下的多线程限流。该算法不保证公平性。除了同步接口外,还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

    RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter");
    // 初始化
    // 最大流速 = 每1秒钟产生10个令牌
    rateLimiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);
    
    CountDownLatch latch = new CountDownLatch(2);
    limiter.acquire(3);
    // ...
    
    Thread t = new Thread(() -> {
        limiter.acquire(2);
        // ...        
    });
    

    相关文章

      网友评论

        本文标题:实操Redission 分布式对象(一)

        本文链接:https://www.haomeiwen.com/subject/ocitphtx.html