Jedis 高级用法实战：事务、管道、Pub/Sub、Token 认证与 UDS 深度指南

Jedis 高级用法实战事务、管道、Pub/Sub、Token 认证与 UDS 深度指南【免费下载链接】jedisRedis Java client项目地址: https://gitcode.com/gh_mirrors/je/jedis导读本文以 JedisRedis Java 客户端的官方高级用法文档为主体深入讲解事务Transaction、管道Pipeline、发布/订阅Pub/Sub、MONITOR 监控、Token 认证、HostAndPortMapper 地址映射以及 Unix Domain SocketUDS连接等进阶能力。读者学完后将掌握如何正确使用Response与exec()编写安全的 Redis 事务如何用管道与集群专用客户端提升批量性能并规避连接池死等如何用JedisPubSub实现消息订阅以及如何通过HostAndPortMapper和JedisSocketFactory打通 NAT / 容器 / 本机高性能等复杂网络场景。文中所引实现均可对照仓库源码src/main/java/redis/clients/jedis进一步验证。Transactions 事务在 Jedis 中执行事务与管道Pipelining非常相似把若干操作包进一个事务块Redis 会保证这些命令作为一个整体被顺序执行且期间不会被其他客户端命令插入。基本用法jedis.watch(key1, key2, ...); Transaction t jedis.multi(); t.set(foo, bar); t.exec();从源码看Transaction继承了AbstractTransaction内部持有一个Connection与一个QueueResponse? pipelinedResponsesTransaction.java。调用jedis.multi()时构造器会把MULTI命令发送给服务端之后所有追加的命令都通过sendCommand排队只有调用exec()时才会一次性发送EXEC并统一取回结果。正因如此事务块内的命令是延迟发送的这一点与管道如出一辙。读取事务内的返回值Response 与 exec()事务中任何有返回值的方法都必须通过ResponseT拿到并在exec()之后再调用Response.get()获取真实结果Transaction t jedis.multi(); t.set(fool, bar); ResponseString result1 t.get(fool); t.zadd(foo, 1, barowitch); t.zadd(foo, 0, barinsky); t.zadd(foo, 0, barikoviev); ResponseSetString sose t.zrange(foo, 0, -1); // get the entire sortedset t.exec(); // dont forget it String foolbar result1.get(); // use Response.get() to retrieve things from a Response int soseSize sose.get().size(); // on sose.get() you can directly call Set methods!需要特别注意Response在exec()之前不包含结果它本质上是一个“Future”式的占位对象。忘记调用exec()会抛出异常。最后一行注释展示的旧式写法ListObject allResults t.exec();依然可用但返回的 List 里混杂着 Redis 的状态消息需要自行提取对象推荐优先使用Response方式。从 exec() 的实现 可以看到exec()会先取回各命令的QUEUED或错误状态再发送EXEC最后把pipelinedResponses中的Response依次set()填充若某条命令排队失败EXECABORT服务端会丢弃整个事务异常中会带上被抑制的错误详情。若发生连接异常事务会被标记为broken true避免在断连状态下继续误操作。事务内禁止使用中间结果Redis 不允许在同一个事务内使用该事务的中间结果下面的代码无法工作// this does not work! Intra-transaction dependencies are not supported by Redis! jedis.watch(...); Transaction t jedis.multi(); if(t.get(key1).equals(something)) t.set(key2, value2); else t.set(key, value);原因在于事务中的命令是排队后一次性执行客户端无法在事务中间“读取”尚未执行的结果。解决这类条件逻辑的官方途径有三个使用本身内置条件语义的原子命令如setnx、getset、incr等它们天然支持“条件执行”使用WATCH 重试模式实现乐观锁WATCH后若键被修改exec()返回null客户端可重试使用EVAL/ Lua 脚本把自定义判断逻辑放到服务端执行。WATCH / UNWATCH / MULTI 的底层协调Transaction提供了doMultifalse的构造器Transaction.java用于手动组合WATCH/UNWATCH/MULTI的调用顺序。默认构造器会自动发送MULTI此时不应再手动调用MULTI。watch()在Connection上直接执行命令并置inWatch true而discard()会发送DISCARD并清空排队响应close()时若未提交则会自动执行discard()或unwatch()清理现场。Pipelining 管道当需要一次性发送大量命令时管道可以显著优于逐条同步发送客户端不必等待每条命令的响应而是全部发出后再统一读取响应从而减少网络往返RTT。基本用法Pipeline p jedis.pipelined(); p.set(fool, bar); p.zadd(foo, 1, barowitch); p.zadd(foo, 0, barinsky); p.zadd(foo, 0, barikoviev); ResponseString pipeString p.get(fool); ResponseSetString sose p.zrange(foo, 0, -1); p.sync(); int soseSize sose.get().size(); SetString setBack sose.get();与事务一致管道的Response也需要在sync()之后读取sync()与syncAndReturnAll()分别对应“只同步结果”与“一次性取回全部结果”两种收尾方式Pipeline.java。集群管道的连接池死等风险与规避Jedis 的高级用法文档特别提醒了一个集群场景下的隐患集群管道会从它命中的每个节点各借用一条连接直到sync()或close()才归还。当多个并发管道以不同顺序访问不同节点时很小的每节点连接池就可能被占满而默认的连接池等待是“无上限”的于是线程会互相无限等待、形成死锁。建议的规避手段尽量让每条管道只访问一个分片single-shard避免同时借用多个节点的连接对高并发多节点管道工作负载使用专用的集群客户端并按预期并发管道数配置每个节点池的大小同时设置有限的最大等待时间maxWait使池耗尽时在确定时间内快速失败而不是无限阻塞。官方推荐写法使用RedisClusterClientClusterPipelineConnectionPoolConfig pipelinePoolConfig new ConnectionPoolConfig(); pipelinePoolConfig.setMaxTotal(expectedConcurrentPipelines); pipelinePoolConfig.setMaxWait(Duration.ofSeconds(1)); try (RedisClusterClient pipelineClient RedisClusterClient.builder() .nodes(nodes) .clientConfig(clientConfig) .poolConfig(pipelinePoolConfig) .build(); ClusterPipeline pipeline pipelineClient.pipelined()) { // Append commands, then call sync() before reading responses. }这里ConnectionPoolConfig对应仓库中的 ConnectionPoolConfig.javaRedisClusterClient与ClusterPipeline分别对应 RedisClusterClient.java 与 ClusterPipeline.java。maxWait的单位在现代版本中支持Duration请以你所依赖的 Jedis 版本 API 为准。Publish/Subscribe 发布订阅Jedis 通过JedisPubSub抽象类实现 Redis 的发布/订阅协议。继承并覆写相应回调方法即可监听消息class MyListener extends JedisPubSub { public void onMessage(String channel, String message) { } public void onSubscribe(String channel, int subscribedChannels) { } public void onUnsubscribe(String channel, int subscribedChannels) { } public void onPSubscribe(String pattern, int subscribedChannels) { } public void onPUnsubscribe(String pattern, int subscribedChannels) { } public void onPMessage(String pattern, String channel, String message) { } } MyListener l new MyListener(); jedis.subscribe(l, foo);关键行为说明subscribe是阻塞操作它会在调用它的线程上持续轮询 Redis 的订阅响应。因此在生产代码中订阅通常应放在独立线程或线程池中执行避免阻塞主业务线程。一个JedisPubSub实例可订阅多个频道在订阅期间调用subscribe/psubscribe可以动态变更订阅集合。回调方法语义onMessage/onSubscribe/onUnsubscribe对应普通频道onPMessage/onPSubscribe/onPUnsubscribe对应模式匹配pattern订阅其中onPMessage额外带回匹配到的pattern参数。类层次上JedisPubSub.java 继承自泛型基类 JedisPubSubBase.java同一套订阅机制也支撑着BinaryJedisPubSub等二进制变体RedisSentinelClient、集群环境等也复用该回调体系。Monitoring 监控Redis 的MONITOR命令可以实时输出服务端收到的每一条命令。Jedis 通过JedisMonitor抽象类暴露回调new Thread(new Runnable() { public void run() { Jedis j new Jedis(localhost); for (int i 0; i 100; i) { j.incr(foobared); try { Thread.sleep(200); } catch (InterruptedException e) { } } j.disconnect(); } }).start(); jedis.monitor(new JedisMonitor() { public void onCommand(String command) { System.out.println(command); } });从 JedisMonitor.java 的源码可以看到proceed(Connection client)会将连接超时设为无限setTimeoutInfinite然后循环读取getBulkReply()并回调onCommand(command)直到连接断开。这意味着monitor同样是一个阻塞调用需要放在独立线程中运行且该连接在监控期间不能复用于普通命令。Token-Based Authentication 基于 Token 的认证从 Jedis 5.3.0 GA 起Jedis 支持基于 Token 的认证机制。仓库中的核心类 AuthXManager.java 实现了SupplierRedisCredentials负责把 Token 认证扩展集成进 Jedis 并驱动整个认证流程提供start()、authenticateConnections(Token)、addConnection(Connection)、stop()等生命周期方法。配套组件还包括 AuthXEventListener.java、TokenCredentials.java 与 JedisAuthenticationException.java。官方文档说明redis-authx-entraid扩展仓库提供了 Jedis 所需组件且对 Microsoft EntraID 的支持已完整落地可作为 Azure Managed RedisAMR与 Azure Cache for RedisACR的认证扩展。使用自定义身份提供方Identity ProviderJedis 提供面向通用身份提供方的 Token 认证机制。你需要自行实现IdentityProvider与IdentityProviderConfig它们来自 Jedis 的传递依赖dependency groupIdredis.clients.authentication/groupId artifactIdredis-authx-core/artifactId version${version}/version /dependency起步示例public class YourCustomIdentityProviderConfig implements IdentityProviderConfig { ... } public class YourCustomIdentityProvider implements IdentityProvider { ... }随后在 Jedis 配置中接入IdentityProviderConfig yourCustomIdentityProviderConfig new YourCustomIdentityProviderConfig(); TokenAuthConfig tokenAuthConfig TokenAuthConfig.builder().identityProviderConfig(yourCustomIdentityProviderConfig); JedisClientConfig config DefaultJedisClientConfig.builder() .authXManager(new AuthXManager(tokenAuthConfig)).build(); ...使用 Microsoft EntraIDEntraID 扩展已与 Azure Managed RedisAMR和 Azure Cache for RedisACR完全集成。添加依赖dependency groupIdredis.clients.authentication/groupId artifactIdredis-authx-entraid/artifactId version${version}/version /dependency然后使用EntraIDTokenAuthConfigBuilder配置... TokenAuthConfig tokenAuthConfig EntraIDTokenAuthConfigBuilder.builder() .expirationRefreshRatio(0.8F) .clientId(yourClientId) .secret(yourClientSecret) .authority(yourAuthority) .scopes(yourRedisScopes).build(); AuthXManager authXManager new AuthXManager(tokenAuthConfig); JedisClientConfig config DefaultJedisClientConfig.builder() .authXManager(authXManager).build(); ...配置项速览配置项含义expirationRefreshRatioToken 过期前的刷新触发比例示例 0.8 表示在剩余有效期 20% 时提前刷新用于平滑续期、避免过期中断clientIdEntraID 应用服务主体的客户端 IDsecret对应的客户端密钥authorityEntraID 租户授权端点scopes申请 Redis 数据面权限所需的 scope这里的AuthXManager是 Jedis 内置类本质上是把 Token 扩展与 Jedis 生命周期连接建立、Token 续期、连接认证粘合在一起。DefaultJedisClientConfig.builder().authXManager(...)的完整构造方式可参考 DefaultJedisClientConfig.java。仓库中另有完整的集成测试用例可供参考例如 TokenBasedAuthenticationIntegrationTests.java、TokenBasedAuthenticationClusterIntegrationTests.java 以及 RedisEntraIDIntegrationTests.java。在实际使用 EntraID 之前还需在 Azure 侧完成 AMR/ACR 服务与 Microsoft EntraID 的配置创建服务主体、授权、配置数据面权限等。此外若你的环境使用自定义身份提供方可参考仓库测试 EntraIDTestContext.java 了解测试上下文的组织方式。HostAndPortMapperNAT / 容器环境下的地址映射在 NAT 网关之后、或 Docker / Kubernetes 等容器编排环境中客户端实际需要连接的地址往往与 Redis Cluster 节点在CLUSTER拓扑里通告的地址不一致例如容器内网 IP vs 宿主机映射端口。Jedis 提供HostAndPortMapper函数式接口来解决这一差异把节点通告的地址动态映射为客户端可达的地址。接口定义非常精简HostAndPortMapper.javaFunctionalInterface public interface HostAndPortMapper { HostAndPort getHostAndPort(HostAndPort hap); }典型场景Docker 端口映射假设 Redis Cluster 运行在远端宿主机的 Docker 中集群配置里节点通告的是容器内地址172.18.0.2:6379 172.18.0.3:6379 172.18.0.4:6379而外部需要通过宿主机 IP 加映射端口访问my-redis.example.com:7001 my-redis.example.com:7002 my-redis.example.com:7003方式一实现专用类适合复杂映射逻辑或需要复用的场景。先定义 mapper 类public class DockerNATMapper implements HostAndPortMapper { // Key: The address reported by Redis (internal). // Value: The address the client should connect to (external). private final MapHostAndPort, HostAndPort mapping; public DockerNATMapper(MapHostAndPort, HostAndPort mapping) { this.mapping mapping; } Override public HostAndPort getHostAndPort(HostAndPort hostAndPort) { return mapping.getOrDefault(hostAndPort, hostAndPort); } }再把它挂到DefaultJedisClientConfig并交给RedisClusterClientMapHostAndPort, HostAndPort nodeMapping new HashMap(); nodeMapping.put(new HostAndPort(172.18.0.2, 6379), new HostAndPort(my-redis.example.com, 7001)); nodeMapping.put(new HostAndPort(172.18.0.3, 6379), new HostAndPort(my-redis.example.com, 7002)); nodeMapping.put(new HostAndPort(172.18.0.4, 6379), new HostAndPort(my-redis.example.com, 7002)); SetHostAndPort initialNodes new HashSet(); // seed node initialNodes.add(new HostAndPort(my-redis.example.com, 7001)); HostAndPortMapper mapper new DockerNATMapper(nodeMapping); JedisClientConfig jedisClientConfig DefaultJedisClientConfig.builder() .user(myuser) .password(mypassword) .hostAndPortMapper(mapper) .build(); RedisClusterClient jedisCluster RedisClusterClient.builder() .nodes(initialNodes) .clientConfig(jedisClientConfig) .build();此后当RedisClusterClient从集群拓扑中发现的节点是172.18.0.2:6379时mapper 会在建连前把它翻译成可访问的地址。方式二Lambda 表达式由于HostAndPortMapper是函数式接口仅一个抽象方法内联映射逻辑可以写得更简洁MapHostAndPort, HostAndPort nodeMapping new HashMap(); nodeMapping.put(new HostAndPort(172.18.0.2, 6379), new HostAndPort(my-redis.example.com, 7001)); nodeMapping.put(new HostAndPort(172.18.0.3, 6379), new HostAndPort(my-redis.example.com, 7002)); nodeMapping.put(new HostAndPort(172.18.0.4, 6379), new HostAndPort(my-redis.example.com, 7002)); SetHostAndPort initialNodes new HashSet(); initialNodes.add(new HostAndPort(my-redis.example.com, 7001)); HostAndPortMapper mapper internalAddress - nodeMapping.getOrDefault(internalAddress, internalAddress); JedisClientConfig jedisClientConfig DefaultJedisClientConfig.builder() .user(myuser) .password(mypassword) .hostAndPortMapper(mapper) .build(); RedisClusterClient jedisCluster RedisClusterClient.builder() .nodes(initialNodes) .clientConfig(jedisClientConfig) .build();底层生效位置从源码看映射发生在建连阶段DefaultJedisSocketFactory持有从JedisClientConfig.getHostAndPortMapper()取到的 mapperDefaultJedisClientConfig.java、DefaultJedisClientConfig.java在getSocketHostAndPort()中调用mapper.getHostAndPort(hap)映射结果非空则替换目标地址DefaultJedisSocketFactory.java。也就是说无论节点地址来自启动时的 seed 节点还是运行期从CLUSTER SLOTS等拓扑发现中得到的地址都会经过 mapper 归一化且未命中映射时默认原样返回。Unix Domain SocketsUDS当客户端与 Redis 服务端位于同一台机器时通过 Unix Domain SocketUDS连接可以绕过 TCP/IP 协议栈获得更低延迟与更高吞吐。Jedis 的做法是实现JedisSocketFactory接口用你偏好的 Unix socket 库如 junixsocket并通过自定义ConnectionProvider交给RedisClient。添加依赖dependency groupIdcom.kohlschutter.junixsocket/groupId artifactIdjunixsocket-core/artifactId version2.10.1/version /dependency实现 UDS Socket Factoryimport org.newsclub.net.unix.AFUNIXSocket; import org.newsclub.net.unix.AFUNIXSocketAddress; public class UdsSocketFactory implements JedisSocketFactory { private final File socketFile; public UdsSocketFactory(String socketPath) { this.socketFile new File(socketPath); } Override public Socket createSocket() throws JedisConnectionException { try { Socket socket AFUNIXSocket.newStrictInstance(); socket.connect(new AFUNIXSocketAddress(socketFile), Protocol.DEFAULT_TIMEOUT); return socket; } catch (IOException e) { throw new JedisConnectionException(Failed to create UDS connection., e); } } }其中JedisSocketFactory接口定义于 JedisSocketFactory.java其createSocket()抛出JedisConnectionExceptionexceptions 包内Protocol.DEFAULT_TIMEOUT是协议层默认超时位于 Protocol.java。通过 RedisClient 连接用自定义 socket factory 创建ConnectionFactory再包进PooledConnectionProviderJedisSocketFactory socketFactory new UdsSocketFactory(/tmp/redis.sock); JedisClientConfig clientConfig DefaultJedisClientConfig.builder().build(); ConnectionFactory connectionFactory new ConnectionFactory(socketFactory, clientConfig); PooledConnectionProvider provider new PooledConnectionProvider(connectionFactory); RedisClient client RedisClient.builder() .connectionProvider(provider) .clientConfig(clientConfig) .build();UDS 之上RedisClient的全部能力照常工作连接池、RESP3 协议、客户端侧缓存client-side caching等均不受影响。相关类型对应仓库中的 ConnectionFactory.java、PooledConnectionProvider.java、RedisClient.java仓库测试 UdsTest.java 提供了 UDS 连接的验证用例。Miscellaneous 杂项要点String 与 Binary什么是“原生”数据Redis 文档强调 String 是基本构建块但这容易造成误解。Redis 的“String”对应 C 语言的char类型8 位与 Java 的String16 位 Unicode并不兼容。Redis 只把数据看作定长的 8 位字节块通常不解释其含义即“binary safe”。因此在 Java 侧byte[]才是“原生”形态可以直接收发String在发送前要经SafeEncoder编码、接收后要解码SafeEncoder.java存在少量性能开销。结论如果你处理的是二进制数据不要先编码成 String直接使用*Binary*系列命令如JedisBinaryCommands、BinaryJedis对应的字节数组重载。关于 Redis 主从master/slave分布的说明一个 Redis 网络由若干 Redis 服务器组成可分为 master 与 slaveslave 与 master 之间通过主从复制保持同步。但对客户端而言master 与 slave 看起来并无区别slave 也接受写请求然而这些写操作不会向上传播且随时可能被 master 的数据覆盖。因此合理的做法是读流量路由到 slave写流量固定发给 master。此外一个 slave 也完全可能被另一个 slave 视为 master即级联复制场景。Jedis 的ReplicaOnlyConnectionResolver/RoundRobinConnectionResolver等连接解析器executors 包即为这类读写分离与多副本调度提供了实现基础。延伸阅读事务与 MULTI 的高级主题参见 docs/transactions-multi.md各类客户端组件的关系与选择参见 docs/redis-client-components-overview.md主从故障切换与 Sentinel参见 docs/failover.md迁移到新版本时的 API 变化参见 docs/migration-guides 目录【免费下载链接】jedisRedis Java client项目地址: https://gitcode.com/gh_mirrors/je/jedis创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考