Redis实战总结-管道和发布、订阅机制

逃离我推掉我的手 2022-06-04 09:27 174阅读 0赞

本篇博客重点介绍Redis的管道，发布/订阅机制。  
Redis是一种基于Client-Server模型以及请求/响应协议的TCP服务。Client端发出请求，server端处理并返回结果到客户端。在这个过程中Client端是以阻塞形式等待服务端的响应。假设从Client发送命令到收到Server的处理结果需要1/16秒，这样带来的结果是Client每秒只能发送16条命令，即使Redis每秒可以处理几百个命令，严重影响了Redis的使用效率。因而针对上述问题Redis实现以下三种机制可以批量式的处理命令（通过减少Client和Server之间的网络通信次数来提升Redis在执行多个命令的性能）。

*  **处理多参数的命令：**比如MGET，MSET，HMGET，HMSET，RPUSH等命令可以接受多个参数，这样能够极大的提升性能，但是这些命令只能处理需要重复执行相同命令的操作。
 *  **Multi和Exec命令：**将多个命令封装成一个事务，以事务的方式提交，然后等待所有回复出现。但是这种方式仍然会消耗资源，并且可能会导致其他重要的命令被延迟执行。正常会和Watch、Unwatch、Discard命令结合使用，确保自己正在使用的数据没有发生变化来避免数据出错。
 *  **管道技术：**本篇博客重点介绍的。

## Redis的管道机制 ##

在Server端未响应时，Client端可以连续式的向Server端发送命令，并最终一次性读取Server端的所有响应，各个命令之间互不干扰。显然这种方式可以显著性地提高了 redis 服务的性能。我们把这种方式称为：非事务性流水线方式。  
下面给出一个具体的测试实例：

/** * 设置给定键的值 * @param key * @param value */
        public static void set(String key, String value) {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.set(key, value);
            } finally {
                safeReturn(jedis);
            }
        }
    
        /** * 测试使用管道的效率 * @param key */
        public static void batchTestPipeLine(String key) {
            Jedis jedis = jedisPool.getResource();
            try {
                Pipeline pipeline = jedis.pipelined();
                for (int i = 0; i < 10000; i++) {
                    Response<Long> returns = pipeline.incr(key);
                }
                pipeline.syncAndReturnAll();
            } finally {
                safeReturn(jedis);
            }
    
        }
    
        /** * 测试不使用管道的效率 * @param key */
        public static void batchTestWithNonePipeLine(String key) {
            Jedis jedis = jedisPool.getResource();
            try {
                for (int i = 0; i < 10000; i++) {
                    jedis.incr(key);
                }
            } finally {
                safeReturn(jedis);
            }
        }

主程序测试：

import org.apache.log4j.Logger;
    import util.JedisUtil;
    
    public class Application2 {
    
        private static Logger logger = Logger.getLogger(Application.class);
    
        private static String PIPELINE_HAS = "PIPELINE_HAS";
        private static String PIPELINE_NONE = "PIPELINE_NONE";
    
        public static void main(String[] args) {
    
            logger.info("主程序开始运行：");
            Long beginTime = System.currentTimeMillis();
    
            JedisUtil.init();
    
            JedisUtil.set(PIPELINE_HAS, "0");
            JedisUtil.set(PIPELINE_NONE, "0");
    
            JedisUtil.batchTestWithNonePipeLine(PIPELINE_NONE);
            Long endTime1 = System.currentTimeMillis();
    
            JedisUtil.batchTestWithNonePipeLine(PIPELINE_HAS);
            Long endTime2 = System.currentTimeMillis();
    
            logger.info("未使用管道技术，消耗时间："+(endTime1-beginTime));
            logger.info("使用管道技术，消耗时间："+(endTime2-endTime1));
        }
    }

程序运行结果：  
![这里写图片描述][SouthEast]  
通过测试发现，使用通道技术和不使用差距还是很明显的。

## Redis发布和订阅机制 ##

Redis的发布与订阅机制由两部分组成：发布者，订阅者。发布者（publisher）负责向频道发送二进制字符串消息；订阅者（subscriber）负责订阅频道。每当发布者将消息发送至给定频道时，频道的所有订阅者都会收到消息。

Redis的发布和订阅是一种消息通信模式，可以解除消息发布者和消息订阅者之间的耦，类似于设计模式中观察者模式。

#### Redis中发布和订阅命令 ####

<table> 
 <thead> 
  <tr> 
   <th align="left">命令</th> 
   <th align="left">用例及描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">SUBSCRIBE</td> 
   <td align="left">SUBSCRIBE channel [channel …]——订阅给定的一个或多个频道</td> 
  </tr> 
  <tr> 
   <td align="left">UNSUBSCRIBE</td> 
   <td align="left">UNSUBSCRIBE [channel [channel …]]——退订给定的一个或多个频道，如果没有给定任何频道，退订所有</td> 
  </tr> 
  <tr> 
   <td align="left">PUBLISH</td> 
   <td align="left">PUBLISH channel message——向给定频道发送消息</td> 
  </tr> 
  <tr> 
   <td align="left">PSUBSCRIBE</td> 
   <td align="left">PSUBSCRIBE pattern [pattern …]——订阅给定模式相匹配的所有频道</td> 
  </tr> 
  <tr> 
   <td align="left">PUNSUBSCRIBE</td> 
   <td align="left">PUNSUBSCRIBE [pattern [pattern …]]——退订给定的模式，如果没有给定任何模式，退订所有</td> 
  </tr> 
 </tbody> 
</table>

#### JAVA中发布和订阅的实现 ####

重写类JedisPubSub中onMessage方法

package listener;
    
    import redis.clients.jedis.JedisPubSub;
    
    import java.util.logging.Logger;
    
    /** * 重写Redis的PUB和SUB方法 * @author guweiyu */
    public class RedisPubSubListener extends JedisPubSub { 
    
        Logger logger = Logger.getLogger(RedisPubSubListener.class.getName());
    
        @Override
        public void onMessage(String channel, String message) {
            logger.info("channel:" + channel + "receives message :" + message);
            this.unsubscribe(channel);
        }
    
        public void unsubscribe(String channel) {
            logger.info("unsubscribe channel:"+channel);
            super.unsubscribe(channel);
        }
    }

/** * 订阅频道 * @param jedisPubSub * @param channel */
        public static void subscribe(JedisPubSub jedisPubSub, String channel) {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.subscribe(jedisPubSub, channel);
            } finally {
                safeReturn(jedis);
            }
        }
    
        /** * 向频道发布信息 * @param channel * @param msg */
        public static void publish(String channel, String msg) {
            Jedis jedis = jedisPool.getResource();
            try {
                jedis.publish(channel, msg);
            } finally {
                safeReturn(jedis);
            }
        }

编写PUB端测试案例

import org.apache.log4j.Logger;
    import util.JedisUtil;
    
    public class ApplicationPub { 
    
        private static Logger logger = Logger.getLogger(ApplicationPub.class);
    
        public static void main(String[] args) throws Exception{
    
            logger.info("Main Application is starting");
            JedisUtil.init();
            JedisUtil.publish("channel_test", "hello, world");
            Thread.sleep(5000);
            JedisUtil.publish("channel_test", "hello, china");
        }
    }

编写SUB端测试案例

import listener.RedisPubSubListener;
    import org.apache.log4j.Logger;
    import util.JedisUtil;
    
    public class ApplicationSub { 
    
        private static Logger logger = Logger.getLogger(ApplicationSub.class);
    
        public static void main(String[] args) {
    
            logger.info("Main Application is starting");
            JedisUtil.init();
    
            // 订阅频道
            RedisPubSubListener listener = new RedisPubSubListener();
            // Redis的subscribe是阻塞式方法，在取消订阅该频道前，会一直阻塞在这
            JedisUtil.subscribe(listener, "channel_test");
            logger.info("订阅结束");
        }
    }

先启动SUB端，再启动PUB端，运行结果如下：  
![这里写图片描述][SouthEast 1]  
![这里写图片描述][SouthEast 2]

在实际中直接使用Redis的PUB和SUB机制处理消息相对较少，由于和数据传输的可靠性有关。如果客户端在执行订阅操作的过程中断线，那么客户端将会丢失在断线期间发送的所有消息。

[SouthEast]: /images/20220604/cb74df332bc54e4ba1afe1d21038576a.png
[SouthEast 1]: /images/20220604/4a0e8eb036b148c0b2f9c116ce98722a.png
[SouthEast 2]: /images/20220604/4218e5c908644f6290e43837bc49d059.png