高考报名时间2022年11月1日还剩 天 距离2023年6月7日高考还有 天
栏目分类:
子分类:
返回
名师互学网用户登录
名师互学网
快速导航关闭
当前搜索
当前分类
前沿技术 软件开发 系统运维 产品运营 生活办公 面试经验 考试题库
子分类
人工智能 大数据 云计算 区块链 物联网 深度学习 机器学习 NLP 计算机视觉 语音识别 其他 大数据系统 数据可视化 数据挖掘与分析 其他 Docker/k8s 虚拟化 云平台 其他 基本原理 数字货币 智能合约 EOS应用 其他 通讯技术 嵌入式开发 单片机 物联网应用 HarmonyOS 其他 后端开发 Web开发 移动开发 游戏开发 Python Java 架构设计 C/C++/C# PHP .Net Go语言 R语言
实用工具
学习工具 小学数学练习 字帖生成 在线画板 函数绘制 拼音字母表 在线词典 黄历查询 亲戚关系计算 安全期计算 中国历史 Excel函数 模拟请求 json格式化 浏览器指纹
热门搜索
路由器设置 木托盘 宝塔面板 儿童python教程 心情低落 朋友圈 vim 双一流学科 专升本 我的学校 日记学校 西点培训学校 汽修学校 情书 化妆学校 塔沟武校 异形模板 西南大学排名 最精辟人生短句 6步教你追回被骗的钱 南昌大学排名 清朝十二帝 北京印刷学院排名 北方工业大学排名 北京航空航天大学排名 首都经济贸易大学排名 中国传媒大学排名 首都师范大学排名 中国地质大学(北京)排名 北京信息科技大学排名
名师互学网 > IT > 软件开发 > 后端开发 > Java

Jedis分析一-Pipeline is currently not supported for JedisClusterConnection.

Java 更新时间： 发布时间： IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

Jedis分析一-Pipeline is currently not supported for JedisClusterConnection.

背景

项目组最近准备将Redis由哨兵模式组网切换到集群组网，切换后应用访问redis时报错，“Pipeline is currently not supported for JedisClusterConnection.”。
初步定为Jedis在集群模式下不支持pipeline。
org.springframework.data.redis.connection.jedis.JedisClusterConnection#openPipeline

	@Override
	public void openPipeline() {
		throw new UnsupportedOperationException("Pipeline is currently not supported for JedisClusterConnection.");
	}
pipeline解析

pipeline并不是redis的设计，只要是基于TCP的长连接，而且访问协议为简单请求/应答，都能去实现pipeline。
比如HTTP中加入keepalive后，就能实现pipeline。而且浏览器也是这么优化的。

如上图所示，客户端一次发送一批命令给服务端，客户端一次解析多个响应结果。
因为有定义应用层协议，无论是请求还是响应，尽管多个报文的字节流挤在一起发送或者接收，可以通过应用层协议去将一帧帧报文解析出来，而不影响各自的处理。
上图为pipeline通用处理过程，具体实现起来，并不严格按照上述流程。比如：

  1. openPipeline、closePileline这两个动作并不是必须的
    对于应用层协议来讲，它并不感知这两个动作。这个仅仅只是客户端代码为了更好的去复用单条指令执行的逻辑而已。openPipeline可以让sdk知道此时是pipeline模式，后续所有的命令都不要发送，等调用closePipeline后一次性发送。
  2. 客户端并不一定是在closePipeline的时候一次性发送所有的指令，这个得看具体的客户端实现，比如Jedis就不是，可以看下面的Jedis是如何实现Pipeline的。
jedis是如何实现Pipline的 RedisTemplate中pipeline的用法 
List result = redisTemplate.executePipelined(new RedisCallback() {
    @Override
    public Object doInRedis(RedisConnection connection) {
        // 这里可以使用connection做任何其支持的操作，但是得到的返回结果都是null
        ...  
        //  doInRedis必须返回null，不然会抛异常，作者估计也是想提醒开发者，这里的返回值没有任何意义   
        return null;
    }
});
 
源码解析 

org.springframework.data.redis.core.RedisTemplate#executePipelined(org.springframework.data.redis.core.RedisCallback, org.springframework.data.redis.serializer.RedisSerializer)
 代码有省略
 

	public List executePipelined(RedisCallback action, @Nullable RedisSerializer resultSerializer) {

		return execute((RedisCallback>) connection -> {
			connection.openPipeline();
			boolean pipelinedClosed = false;
			try {
				Object result = action.doInRedis(connection);
				List closePipeline = connection.closePipeline();
				pipelinedClosed = true;
				return deserializeMixedResults(closePipeline, resultSerializer, hashKeySerializer, hashValueSerializer);
			} finally {
				if (!pipelinedClosed) {
					connection.closePipeline();
				}
			}
		});
	}
 
connection.openPipeline() 

因为JedisClusterConnection不支持openPipeline，因此先分析一下JedisConnection的源码，可以看到仅仅只是创建了一个Pipeline，用于容纳后续响应。
 

org.springframework.data.redis.connection.jedis.JedisConnection#openPipeline
 

	public void openPipeline() {
		if (pipeline == null) {
			pipeline = jedis.pipelined();
		}
	}

 

redis.clients.jedis.BinaryJedis#pipelined
 

  public Pipeline pipelined() {
    pipeline = new Pipeline();
    pipeline.setClient(client);
    return pipeline;
  }

 
action.doInRedis 

action为示例代码中的RedisCallback，我们以最简单的connection.get(“aaa”.getBytes());为例进行分析。
 可以看到代码最终走到了org.springframework.data.redis.connection.jedis.JedisStringCommands#get
 可以看到简单的get命令，会根据是否开启pipeline来决定不同的逻辑。
 

    // 代码有省略
	public byte[] get(byte[] key) {
		if (isPipelined()) {
			pipeline(connection.newJedisResult(connection.getRequiredPipeline().get(key)));
			return null;
		}
		return connection.getJedis().get(key);
	}

 

redis.clients.jedis.PipelineBase#get(byte[])
 

  public Response get(byte[] key) {
    getClient(key).get(key);
    return getResponse(BuilderFactory.BYTE_ARRAY);
  }

 

注意
 可以看到BinaryClient#get方法仅仅只是发送命令，并不会去读取相应。
 另外如前面所说，jedis并不是命令集中起来一次性发送，因为在RedisCallback中调用connection.get(“aaa”.getBytes())时就会走到这里将命令发送出去
 redis.clients.jedis.BinaryClient#get
 

  public void get(final byte[] key) {
    sendCommand(Command.GET, key);
  }

 

可以看到getResponse并不是真正的去读取socket里面的数据去获取响应结果，仅仅只是在pipelinedResponses中放了一个占位符，这样调用connection.closePipeline()时，就知道要获取多少个响应结果。
 redis.clients.jedis.Queable#getResponse
 

  protected  Response getResponse(Builder builder) {
    Response lr = new Response(builder);
    pipelinedResponses.add(lr);
    return lr;
  }

 
connection.closePipeline()解析 

org.springframework.data.redis.connection.jedis.JedisConnection#closePipeline
 该方法比较简单，不做详细介绍，主要逻辑在convertPipelineResults中
 

	public List closePipeline() {
		if (pipeline != null) {
			try {
				return convertPipelineResults();
			} finally {
				pipeline = null;
				pipelinedResults.clear();
			}
		}
		return Collections.emptyList();
	}
 

org.springframework.data.redis.connection.jedis.JedisConnection#convertPipelineResults
 代码有删减，重点在getRequiredPipeline().sync()这一行；这一行代码会去根据pipeline中的命令数去服务端读取对应数量的响应。其他的逻辑仅仅只是判断是否需要对内容做转换不，比如将字符串类型的结果反序列化为对象。
 

	private List convertPipelineResults() {
		List results = new ArrayList<>();
		getRequiredPipeline().sync();
		Exception cause = null;
		for (JedisResult result : pipelinedResults) {
			try {

				Object data = result.get();

				if (!result.isStatus()) {
					results.add(result.conversionRequired() ? result.convert(data) : data);
				}
			}
		}
		return results;
	}
 

redis.clients.jedis.Pipeline#sync
 重点内容在client.getAll();
 

  public void sync() {
    if (getPipelinedResponseLength() > 0) {
      List unformatted = client.getAll();
      for (Object o : unformatted) {
        generateResponse(o);
      }
    }
  }
 

redis.clients.jedis.Connection#getAll()
 从下面源码来看，跟前面写的内容有些出入，调用connection.closePipeline()时，是根据之前那执行sendCommand函数对pipelinedCommands进行的计数，才知道要读取多少个响应结果。
 

public List getAll() {
    return getAll(0);
}

public List getAll(int except) {
    List all = new ArrayList();
    flush();
    while (pipelinedCommands > except) {
      // readProtocolWithCheckingBroken就是根据redis设计的报文格式去读取一次响应的结果
      all.add(readProtocolWithCheckingBroken());
      pipelinedCommands--;
    }
    return all;
}

// 发送命令的时候会对pipelinedCommands进行计数
protected Connection sendCommand(final Command cmd, final byte[]... args) {
    try {
      connect();
      Protocol.sendCommand(outputStream, cmd, args);
      pipelinedCommands++;
      return this;
    }
}
 

至此jedis对pipeline的实现的主要逻辑已经分析清楚，由上可以看到整个pipeline分三步
 

  1. 执行connection.openPipeline开启pipeline，主要就是创建Pipeline这个类，用于后面存放响应的占位符。
  2. 执行回调，回调里面执行的所有命令都会直接发送到服务端，并不会集中起来一起发送。只是发送完命令后，不会立即读取服务端的响应。
  3. 执行connection.closePipeline，根据执行执行的命令计数，去读取n次命令的结果。
 
为啥jedis集群模式下不支持pipeline 

待续







转载请注明：文章转载自 www.mshxw.com















Java相关栏目本月热门文章












热门相关搜索






路由器设置
木托盘
宝塔面板
儿童python教程
心情低落
朋友圈
vim
双一流学科
专升本
我的学校
日记学校
西点培训学校
汽修学校
情书
化妆学校
塔沟武校
异形模板
西南大学排名
最精辟人生短句
6步教你追回被骗的钱
南昌大学排名
清朝十二帝
北京印刷学院排名
北方工业大学排名
北京航空航天大学排名
首都经济贸易大学排名
中国传媒大学排名
首都师范大学排名
中国地质大学(北京)排名
北京信息科技大学排名
中央民族大学排名
北京舞蹈学院排名
北京电影学院排名
中国戏曲学院排名
河北政法职业学院排名
河北经贸大学排名
天津中德应用技术大学排名
天津医学高等专科学校排名
天津美术学院排名
天津音乐学院排名
天津工业大学排名
北京工业大学耿丹学院排名
北京警察学院排名
天津科技大学排名
北京邮电大学(宏福校区)排名
北京网络职业学院排名
北京大学医学部排名
河北科技大学排名
河北地质大学排名
河北体育学院排名















学习工具


代数计算器


三角函数


解析几何


立体几何






知识解答


教育知识


百科知识


生活知识


常识知识






写作必备


作文大全


作文素材


句子大全



实用范文






关于我们


关于我们


联系我们


网站地图






 交流群

名师互学网交流群




名师互学网客服

名师互学网客服












名师互学网 版权所有 (c)2021-2022      ICP备案号：晋ICP备2021003244-6号
 








我们一直用心在做

关于我们
文章归档
网站地图
联系我们

版权所有 (c)2021-2022 MSHXW.COM


ICP备案号：晋ICP备2021003244-6号