import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
@RestController
public class RedisController {
 
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
 
    @GetMapping("/set")
    public String setKey() {
        redisTemplate.opsForValue().set("testKey", "testValue");
        return "Key set successfully";
    }
 
    @GetMapping("/get")
    public Object getKey() {
        return redisTemplate.opsForValue().get("testKey");
    }
}

这段代码展示了如何在Spring Boot应用中使用RedisTemplate操作Redis数据库。setKey方法通过RedisTemplate的opsForValue().set方法设置一个键值对，getKey方法通过opsForValue().get方法获取键对应的值。这个例子简单地展示了如何在Spring Boot中集成Redis，并进行基本的读写操作。

Redis主从复制是一种数据复制的模式，其中一个Redis服务器（主节点）与其他Redis服务器（从节点）进行数据同步。数据是从主节点向从节点同步的，因此从节点只能提供读操作。

一主一从结构：一个主节点和一个从节点。

一主多从结构：一个主节点和多个从节点。

树形结构：主从结构可以通过添加从节点构成树形结构。

配置主从复制：

1. 在从节点的redis.conf文件中添加如下配置：

slaveof <master-ip> <master-port>

2. 如果主节点设置了密码，需要在从节点的redis.conf中添加：

masterauth <master-password>

3. 重启Redis从节点服务使配置生效。

示例：

一主一从结构配置：

在从服务器的redis.conf文件中添加：

slaveof 192.168.1.100 6379

一主多从结构配置：

在每个从服务器的redis.conf文件中添加对应的主服务器IP和端口：

slaveof 192.168.1.100 6379

树形结构配置：

可以通过将一个从节点配置为另一个从节点的主节点来实现。例如，如果有两个从节点A和B，可以将从节点B设置为从节点A的从节点：

在从节点B的redis.conf文件中添加：

slaveof 192.168.1.101 6379

其中192.168.1.101是从节点A的IP，6379是从节点A的端口。

Redis数据迁移通常涉及以下几种方法：

1. 使用SAVE或BGSAVE命令手动生成RDB文件并将其复制到目标服务器。
2. 使用DUMP和RESTORE命令直接在Redis实例间迁移键。
3. 使用redis-cli --pipe来导出数据然后导入到另一个Redis实例。
4. 使用第三方工具如redis-shake或redis-migrate-tool。

以下是使用DUMP和RESTORE命令迁移键的例子：

# 在源Redis服务器上，为要迁移的键使用DUMP命令
redis-cli DUMP mykey > mykey.dump

# 将dump文件传输到目标Redis服务器
scp mykey.dump user@targetserver:/path/to/mykey.dump

# 在目标Redis服务器上，使用RESTORE命令加载键
redis-cli -h targetserver RESTORE mykey 0 /path/to/mykey.dump

请注意，这些方法可能需要额外的步骤，如调整配置文件或设置正确的网络连接，并且在生产环境中执行时应格外小心。

import redis.clients.jedis.Jedis;
import com.fasterxml.jackson.databind.ObjectMapper;
 
// 假设有一个简单的User类
class User {
    public String name;
    public int age;
 
    // 构造函数、getter和setter省略
}
 
public class RedisJsonExample {
    public static void main(String[] args) {
        // 创建Jedis实例连接到Redis服务器
        Jedis jedis = new Jedis("localhost");
 
        // 创建User对象
        User user = new User();
        user.setName("张三");
        user.setAge(30);
 
        // 使用Jackson库将User对象转换为JSON字符串
        ObjectMapper mapper = new ObjectMapper();
        String userJson = mapper.writeValueAsString(user);
 
        // 将JSON数据存储到Redis中
        jedis.set("user:100", userJson);
 
        // 从Redis中读取JSON数据
        String storedUserJson = jedis.get("user:100");
 
        // 将JSON字符串转换回User对象
        User storedUser = mapper.readValue(storedUserJson, User.class);
 
        // 输出结果
        System.out.println("存储前: " + user);
        System.out.println("存储后: " + storedUser);
 
        // 关闭Jedis连接
        jedis.close();
    }
}

这段代码展示了如何在Java中使用Jedis客户端和Jackson库来存储和检索JSON格式的数据。首先创建了一个Jedis实例连接到Redis服务器，然后使用ObjectMapper将一个User对象序列化为JSON字符串，并存储到Redis中的一个键（key）。接着从Redis中读取这个键的值，并将JSON字符串反序列化回User对象。最后，输出存储前后的对象信息，并关闭Jedis连接。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
 
import java.util.Collections;
 
@Service
public class RedisPipelineService {
 
    @Autowired
    private StringRedisTemplate redisTemplate;
 
    public void batchIncrement(String keyPrefix, int start, int end) {
        // 确保在pipeline中使用的key有相同的前缀
        String[] keys = new String[end - start + 1];
        for (int i = start; i <= end; i++) {
            keys[i - start] = keyPrefix + i;
        }
 
        // 使用Lua脚本保证原子性
        String luaScript = 
            "for i = 1, #KEYS do " +
            "   redis.call('INCR', KEYS[i]) " +
            "end";
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(luaScript);
        redisScript.setResultType(Long.class);
 
        // 执行pipeline批量操作
        redisTemplate.executePipelined((pipeline) -> {
            for (String key : keys) {
                pipeline.opsForValue().increment(key); // 自增操作
            }
            pipeline.script().eval(redisScript, Collections.singletonList(keyPrefix + "counter"), keys); // 使用Lua脚本批量自增
        });
    }
}

这段代码演示了如何在Spring Boot应用中使用StringRedisTemplate来执行一个pipeline操作，该操作将一系列键自增，并使用Lua脚本保证操作的原子性。这是一个简化的例子，实际应用中可能需要更复杂的逻辑处理。

以下是一个简化的示例，展示了如何在Spring项目中集成Redis并实现短信登录功能的核心代码。

// 导入Spring和Redis相关依赖的包
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.concurrent.TimeUnit;
 
// 短信登录服务
@Service
public class SmsLoginService {
 
    @Autowired
    private StringRedisTemplate redisTemplate;
 
    // 发送短信验证码并保存到Redis
    public void sendSmsCode(String phoneNumber) {
        // 生成短信验证码
        String smsCode = generateSmsCode();
 
        // 保存验证码到Redis，设置过期时间
        redisTemplate.opsForValue().set(phoneNumber, smsCode, 5, TimeUnit.MINUTES);
 
        // 调用第三方短信服务API发送短信
        // sendSms(phoneNumber, smsCode);
    }
 
    // 验证短信验证码
    public boolean validateSmsCode(String phoneNumber, String smsCode) {
        String cacheCode = redisTemplate.opsForValue().get(phoneNumber);
        return cacheCode != null && cacheCode.equals(smsCode);
    }
 
    // 生成短信验证码的方法（示例）
    private String generateSmsCode() {
        // 实现验证码生成逻辑
        return "123456"; // 示例验证码
    }
}

在这个示例中，我们定义了一个SmsLoginService服务类，它有一个sendSmsCode方法用于发送短信验证码并将其保存到Redis，同时设置了过期时间。还有一个validateSmsCode方法用于验证用户输入的验证码是否与存储在Redis中的验证码匹配。

在实际应用中，你需要实现与短信服务提供商的集成，以发送和接收短信。同时，验证码的生成逻辑也应该更复杂，以满足安全性要求。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
 
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
 
@Component
public class RedisLock {
 
    @Autowired
    private StringRedisTemplate redisTemplate;
 
    @Autowired
    private DefaultRedisScript<Long> lockScript;
 
    @Autowired
    private DefaultRedisScript<String> unlockScript;
 
    public boolean lock(String key, long expireTime) {
        String value = UUID.randomUUID().toString();
        Long result = redisTemplate.execute(lockScript, Collections.singletonList(key), value, String.valueOf(expireTime));
        return result != null && result.equals(1L);
    }
 
    public boolean unlock(String key, String value) {
        String result = redisTemplate.execute(unlockScript, Collections.singletonList(key), value);
        return "OK".equals(result);
    }
}
 
// Lua 脚本用于加锁
String lockScript = "if redis.call('exists', KEYS[1]) == 0 then redis.call('hset', KEYS[1], ARGV[1], 1) redis.call('pexpire', KEYS[1], ARGV[2]) return 1 elseif redis.call('hexists', KEYS[1], ARGV[1]) == 1 then return 0 else return -1 end";
 
// Lua 脚本用于解锁
String unlockScript = "if redis.call('hexists', KEYS[1], ARGV[1]) == 1 then redis.call('hdel', KEYS[1], ARGV[1]) return 'OK' else return 'INVALID_REQUEST' end";

在这个代码实例中，我们使用了Lua脚本来实现Redis分布式锁。这样做的好处是可以保证加锁和解锁的操作是原子的，避免了在解锁之前有其他线程获得锁的情况发生。同时，我们使用了UUID来生成唯一的客户端标识，避免了因为客户端重复请求导致的问题。这个实现是简化版的，但是它展示了如何在Spring Boot应用程序中使用Redis实现分布式锁的基本概念。

Redis是一种开源的内存中数据结构存储系统，可以用作数据库、缓存和消息传递队列。以下是在Windows和Linux上安装Redis的基本步骤。

Windows

1. 下载Redis: 访问Redis官网下载最新的Windows版本。
2. 安装Redis: 解压下载的文件到指定目录。
3. 运行Redis服务器: 打开命令行窗口并导航到Redis目录，运行以下命令：

redis-server.exe redis.windows.conf

Linux (使用Docker容器)

如果你更喜欢在Linux系统上使用Docker来安装Redis，步骤如下：

1. 安装Docker: 在Linux上安装Docker，可以参考Docker官方文档。
2. 运行Redis容器: 使用以下命令运行Redis容器：

docker run --name myredis -d redis

这将创建一个名为myredis的Docker容器，并运行Redis服务。

如果你需要将Redis端口映射到宿主机，可以使用以下命令：

docker run --name myredis -p 6379:6379 -d redis

这会将容器内部的6379端口映射到宿主机的6379端口上。

连接Redis

无论是在Windows还是Linux上，安装完成后，你可以使用redis-cli.exe工具在命令行中连接到Redis服务器。

redis-cli.exe -h 127.0.0.1 -p 6379

在Docker容器中，你可以通过Docker命令进入容器内部进行操作：

docker exec -it myredis /bin/bash
redis-cli

以上步骤和命令提供了在Windows和Linux上安装和使用Redis的基本过程。

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
 
import java.util.concurrent.TimeUnit;
 
public class RedissonLockExample {
 
    // 配置Redisson客户端
    private static RedissonClient redissonClient = Redisson.create(new Config());
 
    public static void main(String[] args) {
        RLock lock = redissonClient.getLock("anyLock");
 
        try {
            // 尝试获取锁，最多等待100秒，锁定之后最多持有锁10秒
            if (lock.tryLock(100, 10, TimeUnit.SECONDS)) {
                try {
                    // 处理业务逻辑
                    System.out.println("Locked and processing...");
 
                    // 模拟长时间操作
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 释放锁
                    System.out.println("Releasing lock...");
                    lock.unlock();
                }
            }
        } finally {
            // 关闭Redisson客户端
            redissonClient.shutdown();
        }
    }
}

这段代码展示了如何在Spring Boot应用中使用Redisson实现分布式锁。首先，我们创建了一个Redisson客户端实例，然后在主方法中获取了一个锁对象。使用tryLock方法尝试获取锁，并在获取锁之后执行业务逻辑。最后，确保释放锁资源并关闭Redisson客户端。

实时计算框架对比和示例代码：

1. Apache Flink

   Flink 是一个分布式流处理和批处理系统。以下是使用Flink读取Kafka数据并写入HBase的简单示例：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.hbase.HBaseSinkFunction;
import org.apache.flink.streaming.connectors.hbase.HBaseTableSource;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
 
public class FlinkHBaseExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);
        DataStream<String> stream = env.addSource(consumer);
 
        stream.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                Put put = new Put(Bytes.toBytes(key));
                put.addColumn(family, qualifier, Bytes.toBytes(value));
                hTable.put(put);
            }
        });
 
        env.execute("Flink HBase Example");
    }
}

2. Apache Kafka

   Kafka 是一个分布式流处理平台。以下是使用Kafka的Java Producer发送消息的示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
 
public class KafkaProducerExample {
    public static void main(String[] args) {
        Producer<String, String> producer = new KafkaProducer<>(properties);
        for (int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<>("topic", Integer.toString(i), "message " + i));
        producer.close();
    }
}

3. Apache HBase

   HBase 是一个分布式的、版本化的、非关系的数据存储系统。以下是使用HBase的Java API进行写入的示例：

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
 
public class HBaseExample {
    public static void main(String[] args) throws Exception {