Lettuce操作redis

Lettuce是一个高性能基于Java编写的Redis驱动框架，底层集成了Project?Reactor提供天然的反应式编程，通信框架集成了Netty使用了非阻塞IO，5.x版本之后融合了JDK1.8的异步编程特性，在保证高性能的同时提供了十分丰富易用的API。本文主要介绍使用lettuce操作redis，使用到的软件版本：Java?1.8.0_191、Redis?5.0.8、lettuce?5.3.1.RELEASE。

1、引入依赖

<dependency>
    <groupId>io.lettuce</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>5.3.1.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.8.0</version>
</dependency>

2、基本操作

package com.shangjack.redis;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.SetArgs;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.api.sync.RedisCommands;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.Future;

/**
 *?使用Lettuce操作redis
 */
public class LettuceBaseCase {
    private static RedisClient client;
    private StatefulRedisConnection<String, String> connection;

????@Before
    public void before() {
        RedisURI redisUri = RedisURI.builder()
                .withHost("10.49.196.10").withPort(6379).withPassword("123456")
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                .build();
        client = RedisClient.create(redisUri);
        connection = client.connect();
    }

????@After
    public void after() {
        connection.close();
        client.shutdown();
    }

    /**
     *?同步操作
     * api和jedis很类型
     */
????@Test
    public void sync() {
        RedisCommands<String, String> commands = connection.sync();
        String result = commands.set("name", "mayun");
        System.out.println(result);

        SetArgs args = SetArgs.Builder.nx().ex(10);
        result = commands.set("age", "30", args);
        System.out.println(result);
    }

    /**
     *?异步操作
     */
????@Test
    public void async() throws Exception {
        RedisAsyncCommands<String, String> commands = connection.async();
        Future<String> future = commands.set("name", "mayun");
        System.out.println(future.get());

        SetArgs args = SetArgs.Builder.nx().ex(10);
        future = commands.set("age", "30", args);
        System.out.println(future.get());
    }

    /**
     *?响应式API
     */
????@Test
    public void reactive() throws Exception {
        RedisReactiveCommands<String, String> commands = connection.reactive();
        Mono<String> result = commands.set("name", "mayun");
        System.out.println(result.block());

        SetArgs args = SetArgs.Builder.nx().ex(10);
        result = commands.set("age", "30", args);
        result.subscribe(value -> System.out.println(value));

        //开启一个事务，先把counter设置为1，再将counter自增1
        commands.multi().doOnSuccess(r -> {
            commands.set("count", "1").doOnNext(System.out::println).subscribe();
            commands.incr("count").doOnNext(c -> System.out.println(c)).subscribe();
        }).flatMap(s -> commands.exec())
                .doOnNext(transactionResult -> System.out.println(transactionResult.wasDiscarded())).subscribe();

        Thread.sleep(1000 * 5);
    }

}

3、高级操作

package com.shangjack.redis;

import io.lettuce.core.ClientOptions;
import io.lettuce.core.ReadFrom;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.cluster.ClusterClientOptions;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.Executions;
import io.lettuce.core.cluster.api.sync.NodeSelection;
import io.lettuce.core.cluster.api.sync.NodeSelectionCommands;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.codec.StringCodec;
import io.lettuce.core.masterreplica.MasterReplica;
import io.lettuce.core.masterreplica.StatefulRedisMasterReplicaConnection;
import io.lettuce.core.resource.ClientResources;
import io.lettuce.core.resource.DefaultClientResources;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.junit.Test;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.*;

/**
 *?高级操作
 */
public class LettuceAdvanceCase {
    /**
     *?主从模式操作
     */
????@Test
    public void masterSlave() {
        //这里只需要配置一个节点的连接信息，不一定需要是主节点的信息，从节点也可以;可以自动发现主从节点
        RedisURI uri = RedisURI.builder().withHost("10.16.60.42").withPort(6379).withPassword("123456").build();
        RedisClient client = RedisClient.create(uri);
        StatefulRedisMasterReplicaConnection<String, String> connection = MasterReplica.connect(client, StringCodec.UTF8, uri);
        //从节点读书数据
        connection.setReadFrom(ReadFrom.REPLICA);

        RedisCommands<String, String> commands = connection.sync();
        commands.set("name", "刘备");
        System.out.println(commands.get("name"));

        connection.close();
        client.shutdown();
    }

    /**
     *?主从模式操作2
     */
????@Test
    public void masterSlave2() {
        List<RedisURI> uris = new ArrayList();
        uris.add(RedisURI.builder().withHost("10.16.60.42").withPort(6379).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.43").withPort(6379).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.44").withPort(6379).withPassword("123456").build());
        RedisClient client = RedisClient.create();
        StatefulRedisMasterReplicaConnection<String, String> connection = MasterReplica.connect(client, StringCodec.UTF8, uris);
        //从节点读书数据
        connection.setReadFrom(ReadFrom.REPLICA);

        RedisCommands<String, String> commands = connection.sync();
        commands.set("name", "张飞");
        System.out.println(commands.get("name"));

        connection.close();
        client.shutdown();
    }

    /**
     *?哨兵模式操作
     */
????@Test
    public void sentinel() {
        List<RedisURI> uris = new ArrayList();
        uris.add(RedisURI.builder().withSentinel("10.16.60.42", 26379).withSentinelMasterId("mymaster").withPassword("123456").build());
        uris.add(RedisURI.builder().withSentinel("10.16.60.43", 26379).withSentinelMasterId("mymaster").withPassword("123456").build());
        uris.add(RedisURI.builder().withSentinel("10.16.60.44", 26379).withSentinelMasterId("mymaster").withPassword("123456").build());

        RedisClient client = RedisClient.create();
        StatefulRedisMasterReplicaConnection<String, String> connection = MasterReplica.connect(client, StringCodec.UTF8, uris);
        //从节点读书数据
        connection.setReadFrom(ReadFrom.REPLICA);

        RedisCommands<String, String> commands = connection.sync();
        commands.set("name", "赵云");
        System.out.println(commands.get("name"));

        connection.close();
        client.shutdown();
    }

    /**
     *?集群操作
     */
????@Test
    public void cluster() {
        Set<RedisURI> uris = new HashSet<>();
        uris.add(RedisURI.builder().withHost("10.16.60.42").withPort(7000).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.42").withPort(7001).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.43").withPort(7000).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.43").withPort(7001).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.44").withPort(7000).withPassword("123456").build());
        uris.add(RedisURI.builder().withHost("10.16.60.44").withPort(7001).withPassword("123456").build());
        RedisClusterClient client = RedisClusterClient.create(uris);
        StatefulRedisClusterConnection<String, String> connection = client.connect();
        RedisAdvancedClusterCommands<String, String> commands = connection.sync();
        commands.set("name", "关羽");
        System.out.println(commands.get("name"));

        //选择从节点,只读
        NodeSelection<String, String> replicas = commands.replicas();
        NodeSelectionCommands<String, String> nodeSelectionCommands = replicas.commands();
        Executions<List<String>> keys = nodeSelectionCommands.keys("*");
        keys.forEach(key -> System.out.println(key));

        connection.close();
        client.shutdown();
    }

    /**
     *?配置客户端资源(ClientResources)及客户端参数(ClientOptions)
     */
????@Test
    public void resourceAndOption() {
        ClientResources resources = DefaultClientResources.builder()
                .ioThreadPoolSize(4) //I/O线程数
                .computationThreadPoolSize(4) //任务线程数
                .build();
        RedisURI redisUri = RedisURI.builder()
                .withHost("10.49.196.10").withPort(6379).withPassword("123456")
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                .build();
        ClientOptions options = ClientOptions.builder()
                .autoReconnect(true)//是否自动重连
                .pingBeforeActivateConnection(true)//连接激活之前是否执行PING命令
                .build();

        RedisClient client = RedisClient.create(resources, redisUri);
        client.setOptions(options);
        StatefulRedisConnection<String, String> connection = client.connect();
        RedisCommands<String, String> commands = connection.sync();
        System.out.println(commands.get("name"));

        connection.close();
        client.shutdown();
        resources.shutdown();
    }

    /**
     *?配置客户端资源(ClientResources)及客户端参数(ClientOptions)
     *?集群
     */
????@Test
    public void resourceAndOption2() {
        ClientResources resources = DefaultClientResources.builder()
                .ioThreadPoolSize(4) //I/O线程数
                .computationThreadPoolSize(4) //任务线程数
                .build();
        //集群地址，配置其中一个即可，不需要配置全
        RedisURI redisUri = RedisURI.builder()
                .withHost("10.16.60.42").withPort(7000).withPassword("123456")
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                .build();
        ClusterClientOptions options = ClusterClientOptions.builder()
                .autoReconnect(true)//是否自动重连
                .pingBeforeActivateConnection(true)//连接激活之前是否执行PING命令
                .validateClusterNodeMembership(true)//是否校验集群节点的成员关系
                .build();

        RedisClusterClient client = RedisClusterClient.create(resources, redisUri);
        client.setOptions(options);
        StatefulRedisClusterConnection<String, String> connection = client.connect();
        RedisAdvancedClusterCommands<String, String> commands = connection.sync();
        System.out.println(commands.get("name"));

        connection.close();
        client.shutdown();
        resources.shutdown();
    }

    /**
     *?连接池
     */
????@Test
    public void pool() throws Exception {
        RedisURI redisUri = RedisURI.builder()
                .withHost("10.49.196.10").withPort(6379).withPassword("123456")
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                .build();
        RedisClient client = RedisClient.create(redisUri);
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        GenericObjectPool<StatefulRedisConnection<String, String>> pool = ConnectionPoolSupport.createGenericObjectPool(client::connect, poolConfig);
        StatefulRedisConnection<String, String> connection = pool.borrowObject();
        RedisCommands<String, String> commands = connection.sync();
        System.out.println(commands.get("name"));

        connection.close();
        pool.close();
        client.shutdown();
    }

}