93、Redis 之 使用连接池管理Redis6.0以上的连接 及 消息的订阅与发布

2023-10-08 15:30
文章标签 使用 连接 redis 管理 发布 连接池 订阅 消息 93 redis6.0 以上

本文主要是介绍93、Redis 之 使用连接池管理Redis6.0以上的连接 及 消息的订阅与发布，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

★ 使用连接池管理Redis连接

从Redis 6.0开始，Redis可支持使用多线程来接收、处理客户端命令，因此应用程序可使用连接池来管理Redis连接。

上一章讲的是创建单个连接来操作redis数据库，这次使用连接池来操作redis数据库

Lettuce连接池 支持需要 Apache Commons Pool2 的支持，需要添加该依赖

接下来即可在程序中通过类似如下代码片段来创建连接池了。
var conf = new GenericObjectPoolConfig<StatefulRedisConnection<String, String>>();

conf.setMaxTotal(20); // 设置连接池中允许的最大连接数

// 创建连接池对象（其中连接由redisClient的connectPubSub方法创建）
pool = ConnectionPoolSupport.createGenericObjectPool(redisClient::connect, conf);

代码演示

创建连接池对象，创建两个消息订阅者和一个消息发布者，然后操作redis数据库

1、添加依赖
在这里插入图片描述

Subscriper 第一个消息订阅者

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
启动这个消息订阅者的程序
在这里插入图片描述

Subscriper 第二个消息订阅者

直接拷贝第一个消息订阅者，然后修改这个消息订阅者只订阅 c2 这个channel 主题
在这里插入图片描述

Publisher 消息发布者

也是拷贝消息订阅者的代码，因为创建连接池对象的代码都是一样的。
这里只需要把消息订阅的方法改成消息发布的方法就可以了，其他代码一样。

在这里插入图片描述

测试：

测试成功
消息发布者成功发布消息
消息订阅者也能接收到各自订阅的channel的消息
用小黑窗测试也没有问题
在这里插入图片描述

完整代码

Subscriper

package cn.ljh.app;import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.ScoredValue;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;import java.time.Duration;//使用 Lettuce ，这个类是消息订阅者
//通过连接池操作redis数据库
public class Subscriper
{private RedisClient redisClient;//连接池pool对象private GenericObjectPool<StatefulRedisPubSubConnection<String, String>> pool;public void init(){//1、定义RedisURIRedisURI uri = RedisURI.builder().withHost("127.0.0.1").withPort(6379)//选择redis 16个数据库中的哪个数据库.withDatabase(0).withPassword(new char[]{'1', '2', '3', '4', '5', '6'}).withTimeout(Duration.ofMinutes(5)).build();//2、创建 RedisClient 客户端this.redisClient = RedisClient.create(uri);//创建连接池的配置对象//GenericObjectPoolConfig<StatefulRedisConnection<String, String>> conf = new GenericObjectPoolConfig<StatefulRedisConnection<String, String>>();var conf = new GenericObjectPoolConfig<StatefulRedisPubSubConnection<String, String>>();//设置连接池允许的最大连接数conf.setMaxTotal(20);//3、创建连接池对象（其中连接由 redisClient 的 connectPubSub 方法创建）pool = ConnectionPoolSupport.createGenericObjectPool(this.redisClient::connectPubSub, conf);}//关闭资源public void closeResource(){//关闭连接池--先开后关this.pool.close();//关闭RedisClient 客户端------最先开的最后关this.redisClient.shutdown();}//订阅消息的方法public void subscribe() throws Exception{//从连接池中取出连接StatefulRedisPubSubConnection<String, String> conn = this.pool.borrowObject();//4、创建 RedisPubSubCommands -- 作用相当与 RedisTemplate 这种，有各种操作redis的方法RedisPubSubCommands cmd = conn.sync();//监听消息：消息到来时，是通过监听器来实现的conn.addListener(new RedisPubSubAdapter<>(){//匿名内部类重写这3个方法：收到消息、订阅主题、取消订阅主题//接收来自普通的channel的消息，就用这个方法（就是没带模式的，比如那些主从、集群模式，点进RedisPubSubAdapter类里面看）//接收消息的方法@Overridepublic void message(String channel, String message){System.err.printf("从 %s 收到消息 : %s\n " , channel , message);}//订阅普通channel激发的方法，//订阅主题的方法--下面有这个订阅的方法cmd.subscribe("c1", "c2");//不太清楚这个 subscribed方法 和 下面的 cmd.subscribe 方法的关联 todo@Overridepublic void subscribed(String channel, long count){System.err.println("完成订阅 ：" + count);}//不订阅普通的channel所使用方法--取消订阅//取消订阅的方法@Overridepublic void unsubscribed(String channel, long count){System.err.println("取消订阅");}});//订阅消息------订阅了 c1 和 c2 这两个主题 channelcmd.subscribe("c1", "c2");}public static void main(String[] args) throws Exception{Subscriper subscriper = new Subscriper();subscriper.init();subscriper.subscribe();//改程序只订阅了60分钟，超过60分钟就程序就退出不订阅了Thread.sleep(600000);//关闭资源subscriper.closeResource();}
}

Subscriper2

package cn.ljh.app;import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;import java.time.Duration;//使用 Lettuce ，这个类是消息订阅者2
//通过连接池操作redis数据库
public class Subscriper2
{private RedisClient redisClient;//连接池pool对象private GenericObjectPool<StatefulRedisPubSubConnection<String, String>> pool;public void init(){//1、定义RedisURIRedisURI uri = RedisURI.builder().withHost("127.0.0.1").withPort(6379)//选择redis 16个数据库中的哪个数据库.withDatabase(0).withPassword(new char[]{'1', '2', '3', '4', '5', '6'}).withTimeout(Duration.ofMinutes(5)).build();//2、创建 RedisClient 客户端this.redisClient = RedisClient.create(uri);//创建连接池的配置对象//GenericObjectPoolConfig<StatefulRedisConnection<String, String>> conf = new GenericObjectPoolConfig<StatefulRedisConnection<String, String>>();var conf = new GenericObjectPoolConfig<StatefulRedisPubSubConnection<String, String>>();//设置连接池允许的最大连接数conf.setMaxTotal(20);//3、创建连接池对象（其中连接由 redisClient 的 connectPubSub 方法创建）pool = ConnectionPoolSupport.createGenericObjectPool(this.redisClient::connectPubSub, conf);}//关闭资源public void closeResource(){//关闭连接池--先开后关this.pool.close();//关闭RedisClient 客户端------最先开的最后关this.redisClient.shutdown();}//订阅消息的方法public void subscribe() throws Exception{//从连接池中取出连接StatefulRedisPubSubConnection<String, String> conn = this.pool.borrowObject();//4、创建 RedisPubSubCommands -- 作用相当与 RedisTemplate 这种，有各种操作redis的方法RedisPubSubCommands cmd = conn.sync();//监听消息：消息到来时，是通过监听器来实现的conn.addListener(new RedisPubSubAdapter<>(){//接收来自普通的channel的消息，就用这个方法（就是没带模式的，比如那些主从、集群模式，点进RedisPubSubAdapter类里面看），@Overridepublic void message(String channel, String message){System.err.printf("从 %s 收到消息 : %s\n " , channel , message);}//订阅普通channel激发的方法，@Overridepublic void subscribed(String channel, long count){System.err.println("完成订阅 ：" + count);}//不订阅普通的channel所使用方法@Overridepublic void unsubscribed(String channel, long count){System.err.println("取消订阅");}});//订阅消息------订阅了 c2 这个主题 channelcmd.subscribe( "c2");}public static void main(String[] args) throws Exception{Subscriper2 subscriper2 = new Subscriper2();subscriper2.init();subscriper2.subscribe();//改程序只订阅了60分钟，超过60分钟就程序就退出不订阅了Thread.sleep(600000);//关闭资源subscriper2.closeResource();}}

Publisher

package cn.ljh.app;import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;import java.time.Duration;//消息发布者//通过连接池操作redis数据库
public class Publisher
{private RedisClient redisClient;//连接池pool对象private GenericObjectPool<StatefulRedisPubSubConnection<String, String>> pool;public void init(){//1、定义RedisURIRedisURI uri = RedisURI.builder().withHost("127.0.0.1").withPort(6379)//选择redis 16个数据库中的哪个数据库.withDatabase(0).withPassword(new char[]{'1', '2', '3', '4', '5', '6'}).withTimeout(Duration.ofMinutes(5)).build();//2、创建 RedisClient 客户端this.redisClient = RedisClient.create(uri);//创建连接池的配置对象//GenericObjectPoolConfig<StatefulRedisConnection<String, String>> conf = new GenericObjectPoolConfig<StatefulRedisConnection<String, String>>();var conf = new GenericObjectPoolConfig<StatefulRedisPubSubConnection<String, String>>();//设置连接池允许的最大连接数conf.setMaxTotal(20);//3、创建连接池对象（其中连接由 redisClient 的 connectPubSub 方法创建）pool = ConnectionPoolSupport.createGenericObjectPool(this.redisClient::connectPubSub, conf);}//关闭资源public void closeResource(){//关闭连接池--先开后关this.pool.close();//关闭RedisClient 客户端------最先开的最后关this.redisClient.shutdown();}//订阅消息的方法public void publish() throws Exception{//从连接池中取出连接StatefulRedisPubSubConnection<String, String> conn = this.pool.borrowObject();//4、创建 RedisPubSubCommands -- 作用相当与 RedisTemplate 这种，有各种操作redis的方法RedisPubSubCommands cmd = conn.sync();//向这两个channel主题各自发布了一条消息cmd.publish("c2","c2 c2 c2 这是一条来自 c2 这个channel 里面的消息");cmd.publish("c1","c1 c1 c1 这是一条来自 c1 这个channel 里面的消息");//关闭资源redisClient.shutdown();}//发送消息，消息发出去，程序就退出了public static void main(String[] args) throws Exception{Publisher subscriper2 = new Publisher();subscriper2.init();subscriper2.publish();subscriper2.closeResource();}}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.ljh</groupId><artifactId>Lettucepool</artifactId><version>1.0.0</version><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><!-- 引入 Lettuce 这个操作redis的框架的依赖 --><dependency><groupId>io.lettuce</groupId><artifactId>lettuce-core</artifactId><version>6.1.4.RELEASE</version></dependency><!-- 创建连接池对象的依赖 --><dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.9.0</version></dependency></dependencies>
</project>

这篇关于93、Redis 之 使用连接池管理Redis6.0以上的连接 及 消息的订阅与发布的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/166268。

相关文章

W外链微信推广短连接怎么做？

W外链微信推广短连接怎么做？

制作微信推广链接的难点分析 一、内容创作难度 制作微信推广链接时，首先需要创作有吸引力的内容。这不仅要求内容本身有趣、有价值，还要能够激起人们的分享欲望。对于许多企业和个人来说，尤其是那些缺乏创意和写作能力的人来说，这是制作微信推广链接的一大难点。 二、精准定位难度 微信用户群体庞大，不同用户的需求和兴趣各异。因此，制作推广链接时需要精准定位目标受众，以便更有效地吸引他们点击并分享链接
阅读更多...
中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
Hadoop数据压缩使用介绍

Hadoop数据压缩使用介绍

一、压缩原则 （1）运算密集型的Job，少用压缩 （2）IO密集型的Job，多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器 2）要在Hadoop中启用压缩，可以配置如下参数
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
使用opencv优化图片（画面变清晰）

使用opencv优化图片（画面变清晰）

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化，使其看起来更清晰，同时保持尺寸不变，通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多，主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器：相机传
阅读更多...
高效+灵活，万博智云全球发布AWS无代理跨云容灾方案！

高效+灵活，万博智云全球发布AWS无代理跨云容灾方案！

摘要 近日，万博智云推出了基于AWS的无代理跨云容灾解决方案，并与拉丁美洲，中东，亚洲的合作伙伴面向全球开展了联合发布。这一方案以AWS应用环境为基础，将HyperBDR平台的高效、灵活和成本效益优势与无代理功能相结合，为全球企业带来实现了更便捷、经济的数据保护。 一、全球联合发布 9月2日，万博智云CEO Michael Wong在线上平台发布AWS无代理跨云容灾解决方案的阐述视频，介绍了
阅读更多...
综合安防管理平台LntonAIServer视频监控汇聚抖动检测算法优势

综合安防管理平台LntonAIServer视频监控汇聚抖动检测算法优势

LntonAIServer视频质量诊断功能中的抖动检测是一个专门针对视频稳定性进行分析的功能。抖动通常是指视频帧之间的不必要运动，这种运动可能是由于摄像机的移动、传输中的错误或编解码问题导致的。抖动检测对于确保视频内容的平滑性和观看体验至关重要。 优势 1. 提高图像质量 - 清晰度提升：减少抖动，提高图像的清晰度和细节表现力，使得监控画面更加真实可信。 - 细节增强：在低光条件下，抖
阅读更多...
pdfmake生成pdf的使用

pdfmake生成pdf的使用

实际项目中有时会有根据填写的表单数据或者其他格式的数据，将数据自动填充到pdf文件中根据固定模板生成pdf文件的需求 文章目录 利用pdfmake生成pdf文件1.下载安装pdfmake第三方包2.封装生成pdf文件的共用配置3.生成pdf文件的文件模板内容4.调用方法生成pdf 利用pdfmake生成pdf文件 1.下载安装pdfmake第三方包 npm i pdfma
阅读更多...
零基础学习Redis(10) -- zset类型命令使用

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合，内部除了存储元素外，还会存储一个score，存储在zset中的元素会按照score的大小升序排列，不同元素的score可以重复，score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2