Redis--微服务中的高级用法

本文深入探讨Redis在微服务架构中的高级应用，包括事务管理、流水线优化、发布订阅机制及Lua脚本增强数据处理能力，旨在提升Redis性能与数据一致性。

发表于 2020/08/06 更新于 2026/09/21

作者 贾永琪

16 分钟阅读

Redis--微服务中的高级用法

文章信息
原文链接：https://jiayq.blog.csdn.net/article/details/107847122
发布时间：2020-08-06 18:45:27
阅读量：758
分类：微服务同时被 3 个专栏收录, 订阅专栏, spring boot, Redis
标签：#Redis在微服务中的高级用法, #微服务中的Redis事务, #微服务中的Redis发布订阅, #微服务中发布订阅源码解析, #微服务中Redis如何提升性能

摘要

文章浏览阅读758次。本文深入探讨Redis在微服务架构中的高级应用，包括事务管理、流水线优化、发布订阅机制及Lua脚本增强数据处理能力，旨在提升Redis性能与数据一致性。

Redis–微服务中的高级用法

Redis 的事务
- redis 事务正常执行
  - redis 事务不会回滚
  - watch 的值不一致不执行
Redis 的流水线
Redis 的发布订阅
Redis 发布订阅源码浅分析
- addListener
  - lazyListen
Java基础–synchronized原理详解
Redis 的lua脚本

Redis 的事务

在spring boot中使用redis事务，和使用redis-cli使用事务是一样的。

Redis事务

redis 事务正常执行

@SpringBootTest
public class MultiRedisTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void testMulti() {
        redisTemplate.opsForValue().set("mu", "3");
        List list = (List) redisTemplate.execute((RedisOperations ro) -> {
           ro.watch("mu");
           ro.multi();
           ro.opsForValue().set("mu1", "mu1");
            System.out.println("mu = " + ro.opsForValue().get("mu"));
            System.out.println("mu1 = " + ro.opsForValue().get("mu1"));
            ro.opsForValue().increment("mu");
            System.out.println("mu ++ = " + ro.opsForValue().get("mu"));
//            ro.opsForValue().increment("mu1");
            return ro.exec();
        });
        System.out.println(list);
    }

}

使用redis-cli验证

redis 事务不会回滚

清空redis:

修改之前的Java代码

最后一步出现异常，redis也不会回滚，最终的结果和正常执行是一样的。

watch 的值不一致不执行

清空rediis,在正常执行的中间，打入断点：

然后开始调试：

此时，redis中还未存入mu1,redis中的mu的值是3.

我们使用redis-cli将mu的值修改为不等于3，此时watch的值发生变化，最终exec不会执行任何命令。

最终mu1也没有存储redis.

Redis 的流水线

Redis是一种基于客户端-服务端模型以及请求/响应协议的TCP服务。

这意味着通常情况下一个请求会遵循以下步骤：

客户端向服务端发送一个查询请求，并监听Socket返回，通常是以阻塞模式，等待服务端响应。
服务端处理命令，并将结果返回给客户端。

客户端和服务器通过网络进行连接。这个连接可以很快（loopback接口）或很慢（建立了一个多次跳转的网络连接）。无论网络延如何延时，数据包总是能从客户端到达服务器，并从服务器返回数据回复客户端。

这个时间被称之为 RTT (Round Trip Time - 往返时间). 当客户端需要在一个批处理中执行多次请求时很容易看到这是如何影响性能的（例如添加许多元素到同一个list，或者用很多Keys填充数据库）。例如，如果RTT时间是250毫秒（在一个很慢的连接下），即使服务器每秒能处理100k的请求数，我们每秒最多也只能处理4个请求。

对于关系数据库中我们可以使用批量，也就是只有需要执行SQL时，才一次性发送全部的SQL，然后执行，这样就能够避免中间因网络等原因造成的网络时间消耗。

在很多情况下并不是Redis性能不佳，而是网络传输的速度造成瓶颈，使用流水线后就可以大幅度的在需要执行很多命令时的Redis性能。

@SpringBootTest
public class PipelineRedisTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void testPipeline() {
        long start = System.currentTimeMillis();
        redisTemplate.executePipelined((RedisOperations ro) -> {
            for (int i = 0; i < 100000; i++) {
                ro.opsForValue().set("key_" + i, i + "");
                if (i % 10000 == 0) {
                    System.out.println("第 " + i + " 个");
                }
            }
            return null;
        });
        long end = System.currentTimeMillis();
        System.out.println(end - start);
    }

}

10W次redis操作，只使用了2.77秒

如果，不使用流水线呢？

@SpringBootTest
public class PipelineRedisTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void testPipeline() {
        long start = System.currentTimeMillis();
//        redisTemplate.executePipelined((RedisOperations ro) -> {
//            for (int i = 0; i < 100000; i++) {
//                ro.opsForValue().set("key_" + i, i + "");
//                if (i % 10000 == 0) {
//                    System.out.println("第 " + i + " 个");
//                }
//            }
//            return null;
//        });
        for (int i = 0; i < 100000; i++) {
            redisTemplate.opsForValue().set("key_" + i, i + "");
            if (i % 10000 == 0) {
                System.out.println("第 " + i + " 个");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println(end - start);
    }

}

可以看到，相同的操作，使用流水线为2770毫秒，不使用流水线，则是使用了94717毫秒。

性能差了34倍

这还是在内网环境下，网络连接的耗时基本可以忽略的那种。

如果是外网环境下，那么网络消耗非常大，那么性能差距会更大。

Redis 的发布订阅

redis在使用redis-cli使用发布订阅，可以实现非常强大的功能。

Redis 发布/订阅

那么，在spring boot 中如何使用发布订阅呢？？

我们简单的使用一个小例子进行验证

@Component
public class Sub1 implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 消息体
        String body = new String(message.getBody());
        // 渠道 -- key/pattern
        String topicName = new String(pattern);

        System.out.println(body);
        System.out.println(topicName);
    }
}

我们首先创建了一个监听器，监听器必须实现MessageListener接口。

监听器接收到消息后，将消息体和对应的topic输出。

@SpringBootApplication(scanBasePackages = "com.study.redishello")
public class RedishelloApplication {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private Sub1 sub1;

    @Autowired
    private ThreadPoolTaskScheduler taskScheduler;

    public static void main(String[] args) {
        SpringApplication.run(RedishelloApplication.class, args);
    }

    @PostConstruct
    public void init(){
        initRedisTemplate();
    }

    public void initRedisTemplate() {
        // RedisTemplate会自动初始化StringRedisSerilaizer，直接可用
        RedisSerializer redisSerializer = redisTemplate.getStringSerializer();
        redisTemplate.setKeySerializer(redisSerializer);
        redisTemplate.setValueSerializer(redisSerializer);
        redisTemplate.setHashKeySerializer(redisSerializer);
        redisTemplate.setHashValueSerializer(redisSerializer);
    }

    @Bean
    public ThreadPoolTaskScheduler initTaskScheduler() {
        if (taskScheduler ==null ) {
            taskScheduler = new ThreadPoolTaskScheduler();
            taskScheduler.setPoolSize(5);
        }
        return taskScheduler;
    }

    @Bean
    public RedisMessageListenerContainer initRedisContainer() {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisTemplate.getConnectionFactory());
        container.setTaskExecutor(taskScheduler);
        Topic topic = new ChannelTopic("topic11");
        container.addMessageListener(sub1, topic);
        return container;
    }
}

在application中，首先创建了一个线程池，线程池用于监听器调度与执行。

然后创建一个监听器容器，将线程池设置给容器。在容器中绑定监听器和topic。

最后启动容器，启动容器后，使用redis-cli发布消息：

此时，在spring boot程序中，阻塞等待topic的消息

我们也可以使用spring boot进行发布消息

@SpringBootTest
public class PubSubRedisTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void testPubSubRedis() {
        redisTemplate.convertAndSend("topic11", "spring_boot");
    }

}

也能正确的监听消息。

Redis 发布订阅源码浅分析

从代码分析，将监听器与topic绑定在一起的，就是这个方法

在RedisMessageListenerContainer中，又会调用两个方法addListener和LazyListen方法

addListener

在内部，有一个Map<MessageListener,Set<Topic>>的map存储监听器与topic的多对多关系。

现在tipic支持两种：

要么是channel，要么是pattern.

如果是channel就存入channelMapping，如果是pattern就存入patternMapping中

如果容器已经处于监听中，那么将本次加入的监听器，也调用底层实现，加入监听：

调用底层客户端，实现监听

这里使用的客户端和我们设置的容器的连接工厂有关。

对于pattern的，使用的底层客户端的psubscribe指令

lazyListen

其中monitor只是单纯的配合synchronized实现加锁

Java基础–synchronized原理详解

我们在初始化容器的时候，设置的是taskExecutor

在RedisMessageListenerContainer中对应的是taskExecutor

但是在属性注入后，会调用afterPropertiesSet，同步设置subscriptionExecutor

接下来还有一个问题，lazyListen中执行的任务是什么结构？

是SubscrpttionTask，这是个什么鬼？

原来SubscriptionTask就是实现了SchedulingAwareRunnable接口的类

这个类到底是干什么的，我其实也不清楚，不过，我们看到实现的接口有一个Runnable，那么，一定和多线程的任务接口Runnable有关：

其内部结构是这样的

在SubscribriptionTask内部有一个PatternSubscriberiptionTask的内部类

线程应该是处于轮询的，每500毫秒询问一次，是否有消息发布。每次轮询3遍。

到目前为止，还剩下最后一个疑问：监听器的onMessage方法是怎么被调用的？

我们前面看到，其真正放到线程池中被执行的是SubscriptionTask

既然SubscriptionTask实现了Runnable接口，而且是放到线程池中执行的，所以，我们应该找SubscriptionTask的run方法。

注意，是SubscriptionTask的run方法，不是PatternSubscribptionTask的run方法。

在run方法中会调用eventuallyPerformSubscription方法在方法中的pSubscribe方法中，创建了一个DispatchMessageListener类

看到了期待的onMessage方法

在DispatchMessageListener的onMessage方法中，调用了dispatchMessage方法

给线程池传入的是一个lamba表达式，具体的操作是processMessage方法

最终就调用到了具体的监听器的onMessage方法了。

最后半部分的调用链想串起来可能不太好找，这里推荐倒着找：

从具体的监听器的onMessage方法入手

查询onMessage的全部调用，然后找RedisMessageListenerContainer的调用

倒着找完，在正着串一次，那么这个调用关系就很明了了。

Java基础–synchronized原理详解

Redis 的lua脚本

Redis中有很多的命令，但是严格来说Redis提供的计算能力还是比较有限的。为了增强Redis的计算能力，Redis在2.6版本后提供了Lua脚本的支持，而且执行Lua脚本在Redis中还具备原子性，所以在需要本证数据一致性的高并发环境中，我们也可以使用Redis的Lua语言来保证数据的一致性，且Lua脚本具备更加强大的运算能力，在高并发需要保证数据一致性时，Lua脚本方案比使用Redis自身提供的事务更加好一些。

在Redis中有两种运行Lua的方法，一种是直接发送Lua到Redis服务器去执行，另一种是先把Lua发送给Redis，Redis会对Lua脚本进行缓存，然后返回一个SHA1的32位编码，之后只需要SHA1和相关参数给Redis便可以执行了。

如果Lua脚本很长，那么就需要通过网络传递脚本执行，实际上，网络速度往往比Redis执行速度慢很多，所以网络就会成为Redis执行的瓶颈。如果可以预先将脚本传输到Redis服务器，在真正调用的时候，只需要像调用方法一样，调用。就能很大程度上提高Redis的执行速度。

RedisScript定义：

接下来我们体验下：

@SpringBootTest
public class LuaRedisTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void testLuaRedis() {
        DefaultRedisScript redisScript = new DefaultRedisScript("return 'hello_lua'");
        redisScript.setResultType(String.class);
        System.out.println(redisTemplate.execute(redisScript, redisTemplate.getStringSerializer(), redisTemplate.getStringSerializer(), null));
    }

}

执行

Redis在微服务中的高级用法, 微服务中的Redis事务, 微服务中Redis如何提升性能

本文由作者按照 CC BY 4.0 进行授权