Redis入门实战(6)-Redisson操作redis

Redisson是一个在Redis的基础上实现的Java常驻内存数据网格(In-Memory Data Grid)。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。Redisson提供了使用Redis的简单、便捷的方法。本文主要介绍使用Redisson操作redis，使用到的软件版本：Java 1.8.0_191、Redis 5.0.8、Redisson 3.13.2。

1、引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.2</version>
</dependency>

2、Redisson操作redis

package com.inspur.demo.general.redis;

import io.reactivex.Scheduler;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.redisson.Redisson;
import org.redisson.api.*;
import org.redisson.config.Config;
import redis.clients.jedis.Jedis;

import java.io.*;
import java.sql.SQLSyntaxErrorException;
import java.util.BitSet;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Redisson操作redis
 * Redisson除了提供同步接口外，还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。
 * Redisson会序列化java对象然后保存到reids，所以通过redis命令行设置的值，Redisson来获取值会报错；通redis命令行获取Redisson设置的值前面会多出序列化相关的信息
 */
public class RedissonCase {
    private RedissonClient client;
    private RedissonReactiveClient reactiveClient;
    private RedissonRxClient rxClient;

    @Before
    public void before() {
        Config config = new Config();
        //config.setCodec(new org.redisson.client.codec.StringCodec());
        config.useSingleServer().setAddress("redis://10.49.196.10:6379").setPassword("123456");
        client = Redisson.create(config);
        reactiveClient = Redisson.createReactive(config);
        rxClient = Redisson.createRx(config);
    }

    @After
    public void after() {
        client.shutdown();
        reactiveClient.shutdown();
        rxClient.shutdown();
    }

    /**
     * 通用对象桶，可以用来存放任类型的对象
     */
    @Test
    public void bucket() throws Exception {
        //同步
        RBucket<String> bucket = client.getBucket("name");
        bucket.set("zhaoyun");
        System.out.println(bucket.get());

        //异步
        RBucket<String> bucket2 = client.getBucket("name2");
        bucket2.setAsync("赵云2").get();
        bucket2.getAsync().thenAccept(System.out::println);

        //Reactive
        RBucketReactive<String> bucket3 = reactiveClient.getBucket("name3");
        bucket3.set("赵云3").block();
        bucket3.get().subscribe(System.out::println);

        //RxJava2
        RBucketRx<String> bucket4 = rxClient.getBucket("name4");
        bucket4.set("赵云4").blockingGet();
        bucket4.get().subscribe(System.out::println);

        Thread.sleep(1000 * 5);
    }

    /**
     * 二进制流
     * 提供了InputStream接口和OutputStream接口的实现
     */
    @Test
    public void stream() throws Exception {
        RBinaryStream stream = client.getBinaryStream("stream");
        stream.set("赵云".getBytes());
        OutputStream outputStream = stream.getOutputStream();
        outputStream.write("张飞".getBytes());

        InputStream inputStream = stream.getInputStream();
        ByteArrayOutputStream result = new ByteArrayOutputStream();
        byte[] b = new byte[1024];
        int len;
        while ((len = inputStream.read(b)) != -1) {
            result.write(b, 0, len);
        }
        System.out.println(result.toString());
    }

    @Test
    public void atomicLong() {
        RAtomicLong atomicLong = client.getAtomicLong("atomicLong");
        atomicLong.set(10);
        atomicLong.incrementAndGet();
        System.out.println(atomicLong);
    }

    /**
     * 限流器
     */
    @Test
    public void rateLimiter() throws InterruptedException {
        RRateLimiter rateLimiter = client.getRateLimiter("rateLimiter");
        //初始化 最大流速:每1秒钟产生5个令牌
        rateLimiter.trySetRate(RateType.OVERALL, 5, 1, RateIntervalUnit.SECONDS);
        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                int i = 0;
                @Override
                public void run() {
                    while(true) {
                        rateLimiter.acquire(1);
                        System.out.println(Thread.currentThread() + "-" + System.currentTimeMillis() + "-" + i++);
                    }
                }
            }).start();
        }

        Thread.sleep(1000 * 5);
    }

    /**
     * RList实现了java.util.List接口
     */
    @Test
    public void list() {
        RList<String> list = client.getList("list");
        list.add("a");
        list.add("赵云");
        list.add("张飞");
        list.remove(1);
        System.out.println(list);
    }

    /**
     * RMap实现了java.util.concurrent.ConcurrentMap接口和java.util.Map接口
     * @throws Exception
     */
    @Test
    public void map() throws Exception {
        RMap<String, String> map = client.getMap("map");
        map.put("name", "赵云");
        map.put("location", "常山");
        map.put("camp", "蜀");
        map.remove("location");
        map.forEach((key, value) -> {System.out.println("key=" + key + ",value=" + value);});
    }

    /**
     * RSet实现了java.util.Set接口
     * @throws Exception
     */
    @Test
    public void set() {
        RSet<String> set = client.getSet("set");
        set.add("赵云");
        set.add("张飞");
        set.forEach(System.out::println);
    }

    /**
     * RQueue实现了java.util.Queue接口
     */
    @Test
    public void queue() {
        RQueue<String> queue = client.getQueue("queue");
        queue.add("赵云");
        queue.add("张飞");
        System.out.println(queue.poll());
        System.out.println(queue.poll());
    }

    /**
     * 可重入锁 RLock实现了java.util.concurrent.locks.Lock接口
     */
    @Test
    public void lock() throws InterruptedException {
        RLock lock = client.getLock("lock");
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread() + "-" + System.currentTimeMillis() + "-" + "获取了锁");
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }).start();
        }
        Thread.sleep(1000 * 5);
    }
}

3、Redisson配置

package com.inspur.demo.general.redis;

import org.junit.Test;
import org.redisson.Redisson;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

/**
 * Redisson配置
 */
public class RedissonConfigCase {
    /**
     * 单Redis节点模式
     */
    @Test
    public void singl() {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://10.49.196.10:6379")
                .setPassword("123456")
                .setConnectionPoolSize(5)//连接池大小
                .setConnectionMinimumIdleSize(2)//最小空闲连接数
                .setDatabase(0);
        RedissonClient client = Redisson.create(config);
        RBucket<String> name = client.getBucket("name");
        System.out.println(name.get());
    }

    /**
     * 主从模式
     */
    @Test
    public void masterSlave() {
        Config config = new Config();
        config.useMasterSlaveServers()
                .setMasterAddress("redis://10.49.196.20:6379")
                .addSlaveAddress("redis://10.49.196.21:6379")
                .addSlaveAddress("redis://10.49.196.22:6379")
                .setPassword("123456")
                .setMasterConnectionPoolSize(5)//主节点连接池大小
                .setMasterConnectionMinimumIdleSize(2)//主节点最小空闲连接数
                .setSlaveConnectionPoolSize(5)//从节点连接池大小
                .setSlaveConnectionMinimumIdleSize(2)//从节点最小空闲连接数
                .setDatabase(0);
        RedissonClient client = Redisson.create(config);
        RBucket<String> name = client.getBucket("name");
        System.out.println(name.get());
    }

    /**
     * 哨兵模式
     */
    @Test
    public void sentinel() {
        Config config = new Config();
        config.useSentinelServers()
                .setMasterName("mymaster")
                .addSentinelAddress("redis://10.49.196.20:26379")
                .addSentinelAddress("redis://10.49.196.21:26379")
                .addSentinelAddress("redis://10.49.196.22:26379")
                .setPassword("123456")
                .setMasterConnectionPoolSize(5)//主节点连接池大小
                .setMasterConnectionMinimumIdleSize(3)//主节点最小空闲连接数
                .setSlaveConnectionPoolSize(5)//从节点连接池大小
                .setSlaveConnectionMinimumIdleSize(3)//从节点最小空闲连接数
                .setCheckSentinelsList(false)
                .setDatabase(0);
        System.out.println(config.useSentinelServers().getSentinelAddresses());

        RedissonClient client = Redisson.create(config);
        RBucket<String> name = client.getBucket("name");
        name.set("赵云");
        System.out.println(name.get());
    }

    /**
     * 集群
     */
    @Test
    public void cluster() {
        Config config = new Config();
        config.useClusterServers()
                .setScanInterval(1000 * 2)
                .addNodeAddress("redis://10.49.196.20:7000", "redis://10.49.196.20:7001")
                .addNodeAddress("redis://10.49.196.21:7000", "redis://10.49.196.21:7001")
                .addNodeAddress("redis://10.49.196.22:7000", "redis://10.49.196.22:7001")
                .setPassword("123456")
                .setMasterConnectionPoolSize(5)
                .setMasterConnectionMinimumIdleSize(2)
                .setSlaveConnectionPoolSize(5)
                .setSlaveConnectionMinimumIdleSize(2);
        RedissonClient client = Redisson.create(config);
        RBucket<String> name = client.getBucket("name");
        name.set("赵云");
        System.out.println(name.get());
    }
}