ZzRG的小站—一个全栈开发爱好者的博客，博客主要用来记录与分享技术、学习技术知识点。
以下均为win系统下的操作：一、创建虚拟环境（1）base环境下操作conda env list #列出所有环境 conda create -n 环境名 python=3.9 #指定 Python 版本，创建虚拟环境conda create --prefix=安装路径\环境名 python=3.9 #指定 Python 版本与安装路径，创建虚拟环境conda remove -n 环境名 --all #删除名为“环境名”的虚拟环境conda activate 环境名 #进入名为“环境名”的虚拟环境（2）虚拟环境内的操作conda list # 列出当前环境下的所有库pip show 包名 # 查看当前环境下某个库的版本 conda deactivate # 退出虚拟环境 （3）路径切换切换上一级目录：cd..切到指定目录的盘符（需到根盘符才能切换成功）： D:切换指定路径：cd D:\Desktop（4）pip安装及相关pip install 包名 http://pypi.douban.com/simple/ --trusted-host http://pypi.douban.com导入：pip install -r requirements.txt导出：pip freeze > requirements.txt安装库的案例：安装指定第三方库：pip install numpy==1.21.5 -i http://pypi.douban.com/simple/ --trusted-host pypi.douban.com（注：网络允许的情况下，直接pip install numpy==1.21.5）（5）conda安装及相关：conda install 包名直接创建虚拟环境及环境：conda env create -f environment.ymal导入：conda install -r requirements.txtconda install --yes --file requirements.txt FOR /F "delims=~" %f in (requirements.txt) DO conda install --yes "%f"导出：conda list -e > requirements.txt注：安装pytorch：建议结合自己电脑CUDA版本，官网安装pytorch安装官网：Previous PyTorch Versions | PyTorch二、关键网址Anaconda官网下载: https://www.anaconda.com/download/pycharm官网下载： https://www.jetbrains.com/pycharm/download/?section=windowspypi国内源及使用帮助: https://mirrors.tuna.tsinghua.edu.cn/help/pypi/​ https://www.cnblogs.com/sunnydou/p/5801760.htmlPytorch官方地址: https://pytorch.org/三、关于.condarc文件以阿里源为例：channels: -defaults show_channel_urls: true default_channels: -http://mirrors.aliyun.com/anaconda/pkgs/main -http://mirrors.aliyun.com/anaconda/pkgs/r -http://mirrors.aliyun.com/anaconda/pkgs/msys2 custom_channels: conda-forge: http://mirrors.aliyun.com/anaconda/cloud msys2: http://mirrors.aliyun.com/anaconda/cloud bioconda: http://mirrors.aliyun.com/anaconda/cloud menpo: http://mirrors.aliyun.com/anaconda/cloud pytorch: http://mirrors.aliyun.com/anaconda/cloud simpleitk: http://mirrors.aliyun.com/anaconda/cloud新建.condarc文件后清理一下：conda clean -i 清华channels: - defaults show_channel_urls: true default_channels: - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free - https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/r custom_channels: conda-forge: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud msys2: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud bioconda: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud menpo: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud pytorch: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud simpleitk: https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud ssl_verify: false
关于字体文件跨域访问限制的解决方法
什么情况下需要跨域请求字体为什么要请求字体？视觉稿中需要使用特别的字体，在带宽允许的情况下，都会选择加载自定义字体来让页面更美观。Iconfont的应用必须加载自定义字体。关于font icon的介绍——http://fontawesome.io/为什么会跨域？CDN加速。为了减少业务服务器压力和加快页面访问速度，大部分的静态资源都会放在CDN上（包括JS、CSS、字体等）。然而CDN域名往往跟主域名不一致。为什么要跨域？这个是规范要求的。https://www.w3.org/TR/css-fonts-3/#font-fetching-requirements解决方案把字体文件放到自己的服务器上。这样就不存在所谓的跨域问题了。把字体文件编码成base64，连同CSS文件一起放到CDN上。因为CSS、JS不会受到同源政策的限制，所以可以绕过跨域问题。不过这里有些建议：字体文件太大就不要这么做，因为CSS文件体积会很大，然后加载半天不出来。一般只适合Iconfont去使用。服务器（CDN）设置。服务器端，在响应头中加入Access-Control-Allow-Origin: *即可。CDN的话，一般有跨域访问限制的配置，把自己的业务网站域名配上去即可。
在使用vue3开发的时候 svg-icon图标不能改颜色！首先为什么svg-icon图标不能改颜色 因为在下载图标的时候自己选了颜色里面有一控制颜色的属性：fill 因此我们在外部使用css如：svg{color：red} 就无法改变颜色了！如何解决呢？1、改变svg-icon图片代码中的fill属性的值（常用）直接在fill属性中添加颜色；fill="#fff"；或者：在内部改为fill=currentColor；就可以使用color属性在css代码中，给svg图片改变颜色了2、在svg标签中定义添加一个fill:currentColor，然后添加color:red，就可以修改svg图标的颜色了.svg{ fill:currentColor; color:red; }
4. gRPC4.1 gRPC介绍1. gRPC是由google开源的一个高性能的RPC框架。Stubby Google内部的RPC,演化而来的，2015正式开源。云原生时代是一个RPC标准。 2. gRPC核心的设计思路 1．网络通信 --->gRPC自己封装网络通信的部分提供多种语言的网络通信的封装(C Java[Netty] Go) 2．协议---> HTTP2传输数据的时候二进制数据内容。支持双向流(双工）连接的多路复用。 3．序列化--->基本文本JSON基于二进制Java原生序列化方式 Thrift二进制的序列化压缩二级制序列化。 protobuf (Protocol Buffers) google开源一种序列化方式︰时间效率和空间效率是JSON的3---5倍。 IDL语言 4．代理的创建--->让调用者像调用本地方法那样去调用远端的服务方法。 stub 3. gRPC 与ThriftRPC区别 共性:支持异构语言的RPC。 区别: 1．网络通信 Thrift TCP 专属协议 GRPC HTTP2 2．性能角度ThriftRPC 性能高于gRPC 3. gRPC 大厂背书(Google)，云原生时代与其他组件合作的顺利。所以gRPC应用更广泛。 4. gRPC的好处 1．高效的进行进程间通信。 2．支持多种语言原生支持C Go Java实现。C语言版本上扩展C++C# NodeJs Python Ruby PHP.. 3．支持多平台运行 Linux Android IOS Mac0s windows。 4. gPRC序列化方式采用protobuf，效率高。 5．使用Http2协议 6．大厂的背书 4.2 Http2.0协议1. 回顾 Http1.x协议 Http1.0协议 请求响应的模式 短链接协议（无状态协议） 传输数据文本结构 单工 无法实现服务的推送 变相实现推送（客户端轮询的方式） Http1.1协议 请求响应的模式 有限的长连接 升级的方式WebSocket 双工 实现服务器向客户端的推送。 管道网络传输(wei'bei's) 总结Http1.x协议 共性 1. 传输数据文本格式 可读性好 但是效率差。 2. 本质上Http1.x协议无法实现双工通信。 3. 资源的请求。需要发送多次请求，建立多个连接才可以完成。 2. HTTP2.0协议 1. http2.0协议是一个二进制协议，效率高于Http1.x协议，可读性差。 2. 可以实现双工通信 3. 一个请求，一个链接 可以请求多个数据。【多路复用】（如下图） 3. Http2.0协议的三个概念 1. 数据liu stream 2. 消息 message 3. 帧 frame 4. 其他的相关概念 1. 数据流的优先级，可以通过为不同的stream设置权重，来限制不同流的传输顺序。 2. 流控 --client发送的数据太快了，server处理不过来，通知client暂停数据的发送。4.3 Protocol buffers[protobuf]1. protobuf 是一种与变成语言无关【IDL】，与集体的平台无关【OS】。 它定义的中间语言，可以方便的在clinet 与server中进行RPC的数据传输。 2. protobuf 两个版本 proto2 proto3，但是目前主流应用都是proto3. 3. protobuf 主要安装protobuf的编译器，编译器的目的，可以把protobf的IDL语言，转换成具体的某一种语言。4.3.1 protobuf的编译器安装IDE 集成4.3.2 protobuf的语法文件格式xxx.proto UserServce.proto ORderService.proto版本设定syntax = "proto3"； (IDL语言)注释1. 单行注释 // 2. 多行注释 /* */与Java语言相关的语法optin 关键字开头的 可以选择使用语法 # 后续protobuf生成的java代码 一个源文件还是多个源文件 xx.java optin java_multiple_files = false; (true为多个或者false一个) 一般情况为false # 指定protobuf生成的类 放置在哪个包中 optin jva_package= "com.suns"; # 指定protobuf生成的外部类的名字（管理内部类【内部类才是正真开发使用】） optin java_outer_classname = "UserName"; (外部类)逻辑包（了解）# 对于protobuf对于文件内容管理 package xxx；导入UserService.proto OrderService.proto import "xxx/UserService.proto";基本类型枚举enm SEASON{ SPRING = 0; SUMMER=1; } 枚举的值 必须是0开始消息 messagesmessages LoginRequset { string username = 1; string password = 2； int32 age = 3； } 编号 从1开始 到2^29-1 注意：19000 -19999 不能用这个区间内的编号，因为他是protobuf自己保留的。 (关键字) - singular ：这个字段的值 只能是0个或1个（默认关键字） null "123456" - repwated message Result{ string conten = 1; repeated string stutas =2; //这个字段 返回值 等价于 java List Protobuf getStatusList() -->List } protobuf[grpc] 可以定义多个消息 message LonginRequest{ .... } message LoginRespone { .... } 消息可以被嵌套 message SerchResponse{ message Result{ string url =1; string titlr = 2; } string xxx =1; int32 yyy =2; Result ppp = 3; } SerchResponse.Result message AAA { string xxx =1; SearchResponse.Result yyy =2; } oneof[其中一个] message SimpleMssage{ onof test_oneof{ string name = 1; int32 age =2; } test_oneof xxx }服务service HelloService{ rpc hello(HelloRequest) returns(HelloResponse){} } # 里面可以定义多个服务方法 # 定义多个服务接口 # gRPC 服务 4个服务方式4.3.3 第一个gRPC的开发项目结构1. xxx-api 模块 定义 protobuf IDL语言 并且通过命令创建具体的代码，后续client server引入使用 1. message 2. service 2. xxx-server 模块 1. 实现api模块中定义的服务接口 2. 发布gPRC服务（创建服务端程序） 3. xxx-clien 模块 1. 创建服务端stub（代理） 2. 基于代理 (stub) PRC调用。api1. .proto文件 书写 protobuf的IDL 2. protoc命令 把proto文件的IDL 转换成编程语言 protoc ---java_out=/xxx/xxx /xxx/xxx/xx.proto 3. 【实战】 mmaven插件 进行protobuf IDL文件的编译，并把它放置IDEA具体位置 pom.xml <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-netty-shaded</artifactId> <version>1.58.0</version> <scope>runtime</scope> </dependency> <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-protobuf</artifactId> <version>1.58.0</version> </dependency> <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-stub</artifactId> <version>1.58.0</version> </dependency> <dependency> <groupId>org.apache.tomcat</groupId> <artifactId>annotations-api</artifactId> <version>6.0.53</version> <scope>provided</scope> </dependency> <build> <extensions> <extension> <groupId>kr.motd.maven</groupId> <artifactId>os-maven-plugin</artifactId> <version>1.7.1</version> </extension> </extensions> <plugins> <plugin> <groupId>org.xolstice.maven.plugins</groupId> <artifactId>protobuf-maven-plugin</artifactId> <version>0.6.1</version> <configuration> <protocArtifact>com.google.protobuf:protoc:3.24.0:exe:${os.detected.classifier}</protocArtifact> <pluginId>grpc-java</pluginId> <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.58.0:exe:${os.detected.classifier}</pluginArtifact> <outputDirectory>${basedir}/src/main/java</outputDirectory> <clearOutputDirectory>false</clearOutputDirectory> </configuration> <executions> <execution> <goals> <goal>compile</goal> <goal>compile-custom</goal> </goals> </execution> </executions> </plugin> </plugins> </build>xxx-server 服务端模块的开发1. 实现业务接口 添加具体的功能 （Mybatis+Mysql） package cn.zzrg.service; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.stub.StreamObserver; /** * @Author ZzRG * @Date 2023/9/24 16:48 * @Version 1.0 */ public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase { /** * 1. 接受client提交的参数 * 2. 业务处理 service + dao 调用对应的业务功能。 * 3. 提供返回值 * @param request * @param responseObserver */ @Override public void hello(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) { // 1. 接收客户端的请求参数 String name = request.getName(); // 2. 业务处理 System.out.println("name parameter" + name); // 3. 封装响应 // 3.1 创建响应对象的构造者 HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder(); // 3.2 填充数据 builder.setResult("hello method invoke ok"); // 3.3 封装响应 HelloProto.HelloResponse helloResponse = builder.build(); responseObserver.onNext(helloResponse); responseObserver.onCompleted(); } } 2. 创建服务端 (Netty) package cn.zzrg.service; import io.grpc.Server; import io.grpc.ServerBuilder; import java.io.IOException; /** * @Author ZzRG * @Date 2023/9/24 16:57 * @Version 1.0 */ public class GrpcServer1 { public static void main(String[] args) throws IOException, InterruptedException { // 1.绑定端口 ServerBuilder<?> serverBuilder = ServerBuilder.forPort(9000); // 2.发布服务 serverBuilder.addService(new HelloServiceImpl()); // 3.创建服务对象 Server server = serverBuilder.build(); server.start(); server.awaitTermination(); } } xxx-client 模块1. client通过代理对象完成远程调用 package cn.zzrg; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; /** * @Author ZzRG * @Date 2023/9/24 17:02 * @Version 1.0 */ public class GrpcClient { public static void main(String[] args) { // 1 创建通信的管道 ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build(); // 2 获得代理对象 stub try { HelloServiceGrpc.HelloServiceBlockingStub helloService = HelloServiceGrpc.newBlockingStub(managedChannel); // 3 完成RPC调用 // 3.1 准备参数 HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder(); builder.setName("ZzRG"); HelloProto.HelloRequest helloRequest = builder.build(); // 3.2 进行功能rpc调用 获取相应的内容 HelloProto.HelloResponse helloResponse = helloService.hello(helloRequest); String result = helloResponse.getResult(); System.out.println("result" + result); } catch (Exception e) { throw new RuntimeException(e); }finally { managedChannel.shutdown(); } } }注意事项服务端 处理返回值时 responseObserver.onNext(helloResponse1); // 通过这个方法 把响应的消息 回传client responseObserver.onCompleted(); //通过client 整个服务结束。 底层返回标记 // clinet就会监听标记[grpc做的] responseObserver.onNext(helloRequest1); responseObserver.onCompleted();4.3.4 gRPC的四种通讯方式四种通讯方式1. 简单rpc 一元rpc（Unary RPC） 2. 服务端流式RPC (Server Streaming RPC) 3. 客户端流式RPC (Client Streaming RPC) 4. 双向流RPC (Bi-directional Stream RPC)简单RPC（一元RPC）1. 第一个RPC程序，实际上就是一元RPC特点当client发起调用后，提交数据，并且等待 服务端响应。 开发过程中，主要采用的就是一元RPC的这种通讯方式语法service HelloService{ rpc hello(HelloRequest) returns (HelloResponse){} rpc hello1(HelloRequest1) returns(HelloResponse1){} }服务端流式RPC一个请求对象，服务端可以回传多个结果对象特点使用场景client ----------> Server 股票的编号 <--------- 某一个时刻的 股票行情proto语法service HelloService{ rpc hello(HelloRequest) returns (stream HelloResponse){} //【服务端流式RPC】 rpc hello1(HelloRequest1) returns(HelloResponse1){} //【一元RPC】 }关键性代码服务端代码： @Override public void c2ss(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) { // 接受client客户端的请求参数 String name = request.getName(); // 做业务处理 System.out.println("name = " + name); // 根据业务处理的结果,提供响应 for (int i = 0; i < 9; i++) { HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder(); builder.setResult("处理结果" + i); HelloProto.HelloResponse helloResponse = builder.build(); responseObserver.onNext(helloResponse); try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } } responseObserver.onCompleted(); } 客户端代码： package cn.zzrg; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import java.util.Iterator; /** * @Author ZzRG * @Date 2023/9/24 17:02 * @Version 1.0 */ public class GrpcClient2 { public static void main(String[] args) { // 获得管道 ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build(); try { HelloServiceGrpc.HelloServiceBlockingStub helloService = HelloServiceGrpc.newBlockingStub(managedChannel); HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder(); builder.setName("ZzRG"); HelloProto.HelloRequest helloRequest = builder.build(); Iterator<HelloProto.HelloResponse> helloResponseIterator = helloService.c2ss(helloRequest); while (helloResponseIterator.hasNext()){ HelloProto.HelloResponse helloResponse = helloResponseIterator.next(); System.out.println("helloResponse.getResult()= " + helloResponse.getResult()); } } catch (Exception e) { throw new RuntimeException(e); }finally { managedChannel.shutdown(); } } } 监听 异步方式 处理服务流式RPC的开发 1. api 2. 服务端 3. 客户端 package cn.zzrg; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.stub.StreamObserver; import java.util.Iterator; import java.util.concurrent.TimeUnit; /** * @Author ZzRG * @Date 2023/9/24 17:02 * @Version 1.0 */ public class GrpcClient3 { public static void main(String[] args) { // 获取管道 ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build(); try { HelloServiceGrpc.HelloServiceStub helloService = HelloServiceGrpc.newStub(managedChannel); HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder(); builder.setName("ZzRG"); HelloProto.HelloRequest helloRequest = builder.build(); helloService.c2ss(helloRequest, new StreamObserver<HelloProto.HelloResponse>() { @Override public void onNext(HelloProto.HelloResponse value) { // 服务端 响应了 一个消息后 需要处理的话 把代码写在这个方法中 System.out.println("服务端每一次响应消息" +value.getResult()); } @Override public void onError(Throwable throwable) { } @Override public void onCompleted() { // 需要把服务 响应的所有数据 拿到后 在进行业务的处理 System.out.println("服务端响应结束 后续可以根据需要 在这里统一处理服务端响应所有的内容"); } }); // 解决相应太快的问题 managedChannel.awaitTermination(5, TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); }finally { // 异步编程因为没有阻塞 顺序执行代码太快了 直接shutdown // 服务端响应 发现client 关了 managedChannel.shutdown(); } } } 客户端流式RPC客户端发送多个请求对象，服务端只返回一个结果应用场景IOT(物联网【传感器】) 向服务端 发送数据proto语法rpc cs2s(stream HelloRequest) returns(HelloResponse){}关键性代码1.api rpc cs2s(stream HelloRequest) returns(HelloResponse){} 2.服务端开发 // 客户端 流式RPC 返回值 @Override public StreamObserver<HelloProto.HelloRequest> cs2s(StreamObserver<HelloProto.HelloResponse> responseObserver) { return new StreamObserver<HelloProto.HelloRequest>() { @Override public void onNext(HelloProto.HelloRequest value) { System.out.println("接受到了client发送的一条消息" + value); } @Override public void onError(Throwable throwable) { } @Override public void onCompleted() { System.out.println("client的所有消息,都发送到了服务器..."); // 提供响应的目的: 当接受了全部的client 提交的信息,并处理后 提供响应 HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder(); builder.setResult("this is result"); HelloProto.HelloResponse helloResponse = builder.build(); responseObserver.onNext(helloResponse); responseObserver.onCompleted(); } }; } 3.客户端开发 package cn.zzrg; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.stub.StreamObserver; import java.util.concurrent.TimeUnit; /** * @Author ZzRG * @Date 2023/9/24 17:02 * @Version 1.0 */ public class GrpcClient4 { public static void main(String[] args) { // 获取管道 ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build(); try { HelloServiceGrpc.HelloServiceStub helloService = HelloServiceGrpc.newStub(managedChannel); // 监听返回响应 StreamObserver<HelloProto.HelloRequest> helloRequestStreamObserver = helloService.cs2s(new StreamObserver<HelloProto.HelloResponse>() { @Override public void onNext(HelloProto.HelloResponse value) { // 监控响应 System.out.println("服务端 响应 数据内容为" + value.getResult()); } @Override public void onError(Throwable throwable) { } @Override public void onCompleted() { System.out.println("服务端响应结束..."); } }); // 客户端 发送数据 到服务端 多条数据 ,不定时... for (int i = 0; i < 10; i++) { HelloProto.HelloRequest.Builder builder = HelloProto.HelloRequest.newBuilder(); builder.setName("ZzRG" + i); HelloProto.HelloRequest helloRequest = builder.build(); helloRequestStreamObserver.onNext(helloRequest); } helloRequestStreamObserver.onCompleted(); // 解决相应太快的问题 managedChannel.awaitTermination(12, TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); }finally { // 异步编程因为没有阻塞 顺序执行代码太快了 直接shutdown // 服务端响应 发现client 关了 managedChannel.shutdown(); } } }双向流式RPC客户端可以发送个请求消息，服务端响应多个响应消息 应用场景聊天室proto语法rpc cs2ss(stream HelloRequest) returns(stream HelloResponse){}关键性代码1. api rpc cs2ss(stream HelloRequest) returns(stream HelloResponse){} 2. 服务端开发 //双向流式RPC的客户端开发 @Override public StreamObserver<HelloProto.HelloRequest> cs2ss(StreamObserver<HelloProto.HelloResponse> responseObserver) { return new StreamObserver<HelloProto.HelloRequest>() { @Override public void onNext(HelloProto.HelloRequest value) { System.out.println("接受到了client发送的一条消息" + value.getName()); responseObserver.onNext(HelloProto.HelloResponse.newBuilder().setResult("response" + value.getName() + "value").build()); } @Override public void onError(Throwable throwable) { } @Override public void onCompleted() { System.out.println("接受到了所有请求的消息...."); responseObserver.onCompleted(); } }; } 3. 客户端开发 package cn.zzrg; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.stub.StreamObserver; import java.util.concurrent.TimeUnit; /** * @Author ZzRG * @Date 2023/9/24 17:02 * @Version 1.0 */ public class GrpcClient5 { public static void main(String[] args) { // 获取管道 ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build(); try { HelloServiceGrpc.HelloServiceStub helloService = HelloServiceGrpc.newStub(managedChannel); // 监听返回响应 StreamObserver<HelloProto.HelloRequest> helloRequestStreamObserver = helloService.cs2s(new StreamObserver<HelloProto.HelloResponse>() { @Override public void onNext(HelloProto.HelloResponse value) { // 监控响应 System.out.println("服务端 响应 数据内容为" + value.getResult()); } @Override public void onError(Throwable throwable) { } @Override public void onCompleted() { System.out.println("服务端响应结束..."); } }); // 客户端 发送数据 到服务端 多条数据 ,不定时... for (int i = 0; i < 10; i++) { HelloProto.HelloRequest helloRequest = HelloProto.HelloRequest.newBuilder().setName("ZzRG" + i).build(); helloRequestStreamObserver.onNext(helloRequest); } helloRequestStreamObserver.onCompleted(); // 解决相应太快的问题 managedChannel.awaitTermination(12, TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); }finally { // 异步编程因为没有阻塞 顺序执行代码太快了 直接shutdown // 服务端响应 发现client 关了 managedChannel.shutdown(); } } }4.3.5 gRPC的代理方式1. BlockingStub 阻塞 通信方式 2. Stub 异步 通过监听处理的 3. FutureStub //不常用 同步 异步 NettyFuture 1. FutureStub 只能应用一元RPC第三种方式package cn.zzrg; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import com.zzrg.TestProto; import com.zzrg.TestServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; /** * @Author ZzRG * @Date 2023/9/24 17:02 * @Version 1.0 */ public class GrpcClient6 { public static void main(String[] args) { // 1 创建通信的管道 ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext().build(); // 2 获得代理对象 stub try { TestServiceGrpc.TestServiceFutureStub testServiceFutureStub = TestServiceGrpc.newFutureStub(managedChannel); ListenableFuture<TestProto.TestResponse> responseListenableFuture = testServiceFutureStub.testZzRG(TestProto.TestRequest.newBuilder().setResult("this is result").build()); // 同步操作 // TestProto.TestResponse testResponse = responseListenableFuture.get(); // System.out.println(testResponse.getName()); // responseListenableFuture.addListener(()->{ // System.out.println("异步的RPC响应 回来了"); // }, Executors.newCachedThreadPool()); Futures.addCallback(responseListenableFuture, new FutureCallback<TestProto.TestResponse>() { @Override public void onSuccess(TestProto.TestResponse result) { System.out.println("result.getResult" + result.getResult()); } @Override public void onFailure(Throwable t) { } }, Executors.newCachedThreadPool()); System.out.println("后续操作..."); managedChannel.awaitTermination(12, TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); }finally { managedChannel.shutdown(); } } }5.gRPC与StringBoot整合5.1gRPC与Stringboot整合的思想1. grpc-server 2. grpc-clientStringBoot与GRPC整个过程中 对于服务端做了什么封装搭建开发环境1. 搭建StringBoot的开发环境 2. 引入GRPC相关内容 <dependency> <groupId>cn.zzrg</groupId> <artifactId>grpc-api</artifactId> <version>1.0-SNAPSHOT</version> </dependency> <dependency> <groupId>net.devh</groupId> <artifactId>grpc-server-spring-boot-starter</artifactId> <version>2.14.0.RELEASE</version> </dependency>开发服务端package com.zzrg.service; import com.zzrg.HelloProto; import com.zzrg.HelloServiceGrpc; import io.grpc.stub.StreamObserver; import net.devh.boot.grpc.server.service.GrpcService; /** * @Author ZzRG * @Date 2023/9/25 12:56 * @Version 1.0 */ // 重复多次 @GrpcService public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase { @Override public void hello(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) { String name = request.getName(); System.out.println("name is" + name); responseObserver.onNext(HelloProto.HelloResponse.newBuilder().setResult("this is result").build()); responseObserver.onCompleted(); } } //yml 核心配置 gRPC端口 spring: application: name: boot-server # 不启动tomcat main: web-application-type: none grpc: server: port: 9000 客户端环境搭建 <dependency> <groupId>net.devh</groupId> <artifactId>grpc-client-spring-boot-starter</artifactId> <version>2.14.0.RELEASE</version> </dependency>编码1. yml spring: application: name: boot-client grpc: client: grpc-server: address: 'static://127.0.0.1:9000' negotiation-type: plaintext 2. 注入stub @GrpcClient("grpc-server") private HelloServiceGrpc.HelloServiceBlockingStub stub; 3. 引入 @RequestMapping("/test") public String test(String name) { System.out.println("name = " + name); HelloProto.HelloResponse helloResponse = stub.hello(HelloProto.HelloRequest.newBuilder().setName(name).build()); return helloResponse.getResult(); }gRPC的高级应用1. 拦截器 一元拦截器 2. Stram Tracer【监听流】 流拦截器 解决后三种通讯方式 3. Retry Policy 客户端重试 4. NameResolver 5. 负载均衡 （pick-first ，轮询） 6. grpc与微服务的整合 序列化 （protobuf）Dobbo grpc Dobbo grpc GateWay grpc JWT grpc Nacos2.0 grpc OpenFeign
Redis为什么快?高性能设计之epoll和IO多路复用深度解析
Redis为什么快?高性能设计之epoll和IO多路复用深度解析IO多路复用解决了哪些问题Redis单线程如何处理那么多并发客户端连接，为什么单线程，为什么快Redis的IO多路复用Redis利用epoll来实现IO多路复用，将连接信息和事件放到队列中，一次放到文件事件分派器，事件分派器将事件分发给事件处理器。Redis是跑在单线程中的，所有的操作都是按照额序线性执行的，但是由于读写操作等待用户输入或输出都是阻塞的，所以I/O操作在一般情况下往往不能直接返回，这会导致某一文件的T/O阻塞导致整个进程无法对其它客户提供服务，而I/O多路复用就是为了解决这个问题而出现所谓I/0多路复用机制，就是说通过一种机制，可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或写就绪），能够通知程序进行相应的读写操作。这种机制的使用需要select 、poll、epoll来配合。多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象上等待，无需阻塞等待所有连接。当某条连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理。Redis服务采用Reactor的方式来实现文件事件处理器（每一个网络连接其实都对应一个文件描述符)Redis基于Reactor模式开发了网络事件处理器，这个处理器被称为文件事件处理器。它的组成结构为4部分:多个套接字、IO多路复用程序、文件事件分派器、事件处理器。因为文件事件分派器队列的消费是单线程的，所以Redis才叫单线程模型BIO当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段:准备数据（对于网络IO来说，很多时候数据在一开始还没有到达。比如，还没有收到一个完整的UDP包。这个时候kernel就要等待足够的数据到来）。这个过程需要等待，也就是说数据被拷贝到操作系统内核的缓冲区中是需要一个过程的。而在用户进程这边，整个进程会被阻塞（当然，是进程自己选择的阻塞)。当kernel一直等到数据准备好了，它就会将数据从kernel中拷贝到用户内存，然后kernel返回结果，用户进程才解除block的状态，重新运行起来。所以，BIO的特点就是在TO执行的两个阶段都被block了。accept 阶段会阻塞 等待客户端连接read 读取阶段也会阻塞在阻塞式I/O模型中，应用程序在从调用recvfrom开始到它返回有数据报准备好这段时间是阻塞的，recvfrom返回成功后，应用进程才能开始处理数据报。存在的问题上面的模型存在很大的问题，如果客户端与服务端建立了连接，如果这个连接的客户端迟迟不发数据，程就会一直堵塞在read()方法上，这样其他客户端也不能进行连接，也就是一次只能处理一个客户端，对客户很不友好解决方法只要连接了一个socket，操作系统分配一个线程来处理，这样read()方法堵塞在每个具体线程上而不堵塞主线程，就能操作多个socket了，哪个线程中的socket有数据，就读哪个socket，各取所需，灵活统一。程序服务端只负责监听是否有客户端连接，使用accept()阻塞客户端1连接服务端，就开辟一个线程(thread1）来执行read()方法，程序服务端继续监听客户端2连接服务端，也开辟一个线程〈thread2）来执行read()方法，程序服务端继续监听客户端3连接服务端，也开辟一个线程（thread3）来执行read()方法，程序服务端继续监听任何一个线程上的socket有数据发送过来，read()就能立马读到，cpu就能进行处理。多线程模型存在的问题每来一个客户端，就要开辟一个线程，如果来1万个客户端，那就要开辟1万个线程。在操作系统中用户态不能直接开辟线程，需要调用内核来创建的一个线程，这其中还涉及到用户状态的切换（上下文的切换），十分耗资源。解决方法第一个办法:使用线程池这个在客户端连接少的情况下可以使用，但是用户量大的情况下，你不知道线程池要多大，太大了内存可能不够，也不可行。第二个办法:NIO（非阻塞式IO）方式因为read()方法堵塞了，所有要开辟多个线程，如果什么方法能使read()方法不堵塞，这样就不用开辟多个线程了，这就用到了另一个IO模型，NIO（非阻塞式IO)NIO当用户进程发出read操作时，如果kernel中的数据还没有准备好，那么它并不会block用户进程，而是立刻返回一个error。从用户进程角度讲，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。用户进程判断结果是一个error时，E就知道数据还没有准备好，于是它可以再次发送read操作。一旦kernel中的数据准备好了，并且又再次收到了用户进程的system call，那么它马上就将数据拷贝到了用户内存，然后返回。所以，NIO特点是用户进程需要不断的主动询问内核数据准备好了吗?一句话，用轮询替代阻塞!在NIO模式中，一切都是非阻塞的:accept()方法是非阻塞的，如果没有客户端连接，就返回无连接标识read()方法是非阻塞的，如果read()方法读取不到数据就返回空闲中标识，如果读取到数据时只阻塞read()方法读数据的时间在NIO模式中，只有一个线程:当一个客户端与服务端进行连接，这个socket就会加入到一个数组中，隔一段时间遍历一次，看这个socket的read()方法能否读到数据，这样一个线程就能处理多个客户端的连接和读取了存在的问题和优缺点NIO成功的解决了BIO需要开启多线程的问题，NIO中一个线程就能解决多个socket，但是还存在2个问题。问题一:这个模型在客户端少的时候十分好用，但是客户端如果很多，比如有1万个客户端进行连接，那么每次循环就要遍历1万个socket，如果一万个socket中只有10个socket有数据，也会遍历一万个socket，就会做很多无用功，每次遍历遇到read返回-1时仍然是一次浪费资源的系统调用。问题二:而且这个遍历过程是在用户态进行的，用户态判断socket是否有数据还是调用内核的read()方法实现的，这就涉及到用户态和内核态的切换，每遍历一个就要切换次，开销很大因为这些问题的存在。优点:不会阻塞在内核的等待数据过程，每次发起的I/О 请求可以立即返回，不用阻塞等待，实时性较好。缺点:轮询将会不断地询问内核，这将占用大量的 CPU时间，系统资源利用率较低，所以一般Web服务器不使用这种T/О模型。结论:让iLinux内核搞定上述需求，我们将一批文件描述符通过一次系统调用传给内核由内核层去遍历，才能真正解决这个问题。IO多路复用应运而生，也即将上述工作直接放进Linux内核，不再两态转换而是直接从内核获得结果，因为内核是非阻塞的。IO多路复用I/0 multiplexing这里面的 multiplexing 指的其实是在单个线程通过记录跟踪每一个Sock(I/O流)的状态来同时管理多个I/O流．目的是尽量多的提高服务器的吞吐能力。大家都用过nginx，nginx使HepolI接收请求，ngnix会有很多链接进来,epoll会把他们都监视起来，然后像拨开关一样，谁有数据就拨向谁，然后调用相应的代码处理。redis类似同理I0 multiplexing就是我们说的select，poll，epoll，有些技术书籍也称这种I0方式为event driven IO事件驱动IO。就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。可以基于一个阻塞对象并同时在多个描述符上等待就绪，而不是使用多个线程(每个文件描述符一个线程，每次new一个线程)，这样可以大大节省系统资源。所以，I/0多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符而这些文件描述符（套按字描述符）其中的任意一个进入读就绪状态, select.poll，epoll等函数就可以返回。Reactor基于I/0复用模型:多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象上等待，无需阻塞等待所有连接。当某条连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理。Reactor模式，是指通过一个或多个输入同时传递给服务处理器的服务请求的事件驱动处理模式。服务端程序处理传入多路请求，并将它们同步分派给请求对应的处理线程，Reactor模式也叫Dispatcher模式。即T/O多了复用统一监听事件，收到事件后分发(Dispatch给某进程)，是编写高性能网络服务器的必备技术。Reactor模式中有2个关键组成:1)Reactor: Reactor在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对IO事件做出反应。它就像公司的电话接线员，它接听来自客户的电话并将线路转移到适当的联系人;2)Handlers:处理租序执行Ⅳ/o事件要完成的实际事件，类似于客户想要与之交谈的公司中的实际办理人。Reactor通过调度适当的处理程序来响应I/O事件，处理程序执行非阻塞操作。select, poll, epoll都是I/O多路复用的具体的实现select方式简单说，就是我们把需要处理的数据封装成FD，然后在用户态时创建一个fd的集合（这个集合的大小是要监听的那个FD的最大值+1，但是大小整体是有限制的 ），这个集合的长度大小是有限制的，同时在这个集合中，标明出来我们要控制哪些数据，比如要监听的数据，是1,2,5三个数据，此时会执行select函数，然后将整个fd发给内核态，内核态会去遍历用户态传递过来的数据，如果发现这里边都数据都没有就绪，就休眠，直到有数据准备好时，就会被唤醒，唤醒之后，再次遍历一遍，看看谁准备好了，然后再将处理掉没有准备好的数据，最后再将这个FD集合写回到用户态中去，此时用户态就知道了，奥，有人准备好了，但是对于用户态而言，并不知道谁处理好了，所以用户态也需要去进行遍历，然后找到对应准备好数据的节点，再去发起读请求，我们会发现，这种模式下他虽然比阻塞IO和非阻塞IO好，但是依然有些麻烦的事情， 比如说频繁的传递fd集合，频繁的去遍历FD等问题select函数的缺点1.bitmap默认大小为1024，虽然可以调整但还是有限度的2.rset每次循环都必须重新置位为0，不可重复使用3.尽管将rset从用户态拷贝到内核态由内核态判断是否有数据，但是还是有拷贝的开销4.当有数据时select就会返回，但是select函数并不知道哪个文件描述符有数据了，后面还需要再次对哀件描述符数组进行遍历。效率比较低poll方式poll的执行流程:1.将五个fd从用户态拷贝到内核态poll为阻塞方法，执行poll方法，如果有数据会将fd对应的revents置为POLLINpoll方法返回循环遍历，查找哪个fd被置位为POLLIN了将revents重置为0便于复用对置位的fd进行读取和处理解决的问题:1.解决了bitmap大小限制2.解决了rset不可重用的情况后面由于二者原理相同，所以没能解决优点1、poll使用pollfd数组来代替select中的bitmap，数组没有1024的限制，可以一次管理更多的client。它和 select的主要区别就是，去掉了select只能监听1024个文件描述符的限制。2、当pollfds数组中有事件发生，相应的revents置位为1，遍历的时候又置位回零，实现pollfd数组的重用epoll方式int epoll_create(int size)；//创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大 int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)； int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout); 1. int epoll_create(int size); 创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大，这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值，参数size并不是限制了epoll所能监听的描述符最大个数，只是对内核初始分配内部数据结构的一个建议。 当创建好epoll句柄后，它就会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。 2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)； 函数是对指定描述符fd执行op操作。 - epfd：是epoll_create()的返回值。 - op：表示op操作，用三个宏来表示：添加EPOLL_CTL_ADD，删除EPOLL_CTL_DEL，修改EPOLL_CTL_MOD。分别添加、删除和修改对fd的监听事件。 - fd：是需要监听的fd（文件描述符） - epoll_event：是告诉内核需要监听什么事，struct epoll_event结构如下： struct epoll_event { __uint32_t events; /* Epoll events */ epoll_data_t data; /* User data variable */ }; //events可以是以下几个宏的集合： EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）； EPOLLOUT：表示对应的文件描述符可以写； EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）； EPOLLERR：表示对应的文件描述符发生错误； EPOLLHUP：表示对应的文件描述符被挂断； EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。 EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里 3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout); 等待epfd上的io事件，最多返回maxevents个事件。 参数events用来从内核得到事件的集合，maxevents告之内核这个events有多大，这个maxevents的值不能大于创建epoll_create()时的size，参数timeout是超时时间（毫秒，0会立即返回，-1将不确定，也有说法说是永久阻塞）。该函数返回需要处理的事件数目，如返回0表示已超时。epoll是非阻塞的是非阻塞的!!epoll的执行流程:1.当有数据的时候，会把相应的文件描述符“置位”，但是epool没有revent标志位，所以并不是真正的置位。这时候会把有数据的文件描述符放到队首。2.epoll会返回有数据的文件描述符的个数3.根据返回的个数读取前N个文件描述符即可4.读取、处理epoll模式是对select和poll的改进，它提供了三个函数：第一个是：eventpoll的函数，他内部包含两个东西一个是：1、红黑树-> 记录的事要监听的FD2、一个是链表->一个链表，记录的是就绪的FD紧接着调用epoll_ctl操作，将要监听的数据添加到红黑树上去，并且给每个fd设置一个监听函数，这个函数会在fd数据就绪时触发，就是准备好了，现在就把fd把数据添加到list_head中去3、调用epoll_wait函数就去等待，在用户态创建一个空的events数组，当就绪之后，我们的回调函数会把数据添加到list_head中去，当调用这个函数的时候，会去检查list_head，当然这个过程需要参考配置的等待时间，可以等一定时间，也可以一直等， 如果在此过程中，检查到了list_head中有数据会将数据添加到链表中，此时将数据放入到events数组中，并且返回对应的操作的数量，用户态的此时收到响应后，从events中拿到对应准备好的数据的节点，再去调用方法去拿数据。多路复用快的原因在于，操作系统提供了这样的系统调用，使得原来的 while循环里多次系统调用变成了一次系统调用＋内核层遍历这些文件描述符。epoll是现在最先进的IO多路复用器，Redis、Nginx，linux中的Java NIO都使用的是epoll。这里“多路”指的是多个网络连接，“复用”指的是复用同一个线程。1、一个socket的生命周期中只有一次从用户态拷贝到内核态的过程，开销小2、使用event事件通知机制，每次socket中有数据会主动通知内核，并加入到就绪链表中，不需要遍历所有的socket总结select模式存在的三个问题：能监听的FD最大不超过1024每次select都需要把所有要监听的FD都拷贝到内核空间每次都要遍历所有FD来判断就绪状态poll模式的问题：poll利用链表解决了select中监听FD上限的问题，但依然要遍历所有FD，如果监听较多，性能会下降epoll模式中如何解决这些问题的？基于epoll实例中的红黑树保存要监听的FD，理论上无上限，而且增删改查效率都非常高每个FD只需要执行一次epoll_ctl添加到红黑树，以后每次epol_wait无需传递任何参数，无需重复拷贝FD到内核空间利用ep_poll_callback机制来监听FD状态，无需遍历所有FD，因此性能不会随监听的FD数量增多而下降
Redis经典五大类型源码及底层实现面试常问的Redis对象object.c字符串t_string.c列表t_list.c字典t_hash.c集合及有序集合t_set.c和t_zset.c数据流t_stream.cRedis基本数据结构（骨架）简单动态字符串sds.c整数集合intset.c压缩列表ziplist.c快速链表quicklist.clistpack字典dict.cRedis数据库的实现数据库的底层实现db.c持久化rdb.c和aof.cRedis服务端和客户端事件驱动ae.c和ae_epoll.c网络连接anet.c和networking.c服务端程序server.c客户端程序redis-cli.c其他主从复制replication.c哨兵sentinel.c集群cluster.c其他数据结构，如hyperloglog.c、geo.c等其他功能，如pub/sub、Lua脚本源码分析Redis中每个对象都是一个redisObject结构每个键值对都会有一个digtEntry(源码位置:dict.h)重点:从dictEntry到RedisObjectZ经典5大数据结构解析//edis结构 //*redis6相关的底层模型和结构*string = sDs *set = intset + hashtable *zset = skipList + zipList *list = quickList + zipList *hash = hashtable + zipList * =================================== * *redis7相关的底层模型和结构 *string = SDS *set = intset + hashtable *zset = skipList + listpack紧凑列表 *List = quicklist *Hash = hashtable + Listpack紧凑列表 */ debug 的开启String数据结构介绍SDS简单动态字符串Hash数据结构介绍redis6hash-max-ziplist-entries:使用压缩列表保存时哈希集合中的最大元素个数。hash-max-ziplist-value:使用压缩列表保存时哈希集合中单个元素的最大长度。结论1.哈希对象保存的键值对数量小于512个;2.所有的键值对的健和值的字符串长度都小于等于64byte (一个英文字母一个字节)时用ziplist，反之用hashtableziplist升级到hashtable可以，反过来降级不可以（一旦从压缩列表转为了哈希表，Hash类型就会一直用哈希表进行保存而不会再转回压缩列表了。在节省内存空间方面哈希表就没有压缩列表高效了。）源码分析hset命令解读ziplist.cziplist为了节省内存，采用了紧凑的连续存储。ziplist是一个双向链表，可以在时间复杂度为O(1）下从头部、尾部进行 pop或push。新增或更新元素可能会出现连锁更新现象(致命缺点导致被listpack替换)。不能保存过多的元素，否则查询效率就会降低，数量小和内容小的情况下可以使用。redis7源码明明有ziplist了，为什么出来一个listpack紧凑列表?Ziplsit的连锁更新问题结论：listpack 是Redis 设计用来取代掉ziplist 的数据结构，它通过每个节点记录自己的长度且放在节点的尾部，来彻底解决掉了ziplist存在的连锁更新的问题ziplist内存布局VS listpack内存布局List数据结构介绍Set数据结构介绍ZSet数据结构介绍总结skiplist跳表跳表是可以实现二分查找的有序链表skiplist是一种以空间换取时间的结构。由于链表，无法进行二分查找，因此借鉴数据库索引的思想，提取出链表中关键节点(索引)，先在关键节点上查找，再进入下层链表查找，提取多层关键节点，就形成了跳跃表but由于索引也要占据一定空间的，所以，索引添加的越多，空间占用的越多总结来讲跳表=链表＋多级索引跳表时间+空间复杂度介绍
Redis的缓存过期淘汰策略面试题生产上你们的redis内存设置多少?如何配置、修改redis的内存大小如果内存满了你怎么办redis默认内存多少?在哪里查看?如何设置修改?内存满了会报 oom往redis里面写的数据是怎么没了的？它是如何删除的？redis过期键的删除策略如果一个键是过期的，那它到了过期时间之后是不是马上就从内存中被被删除呢? ?如果回答yes，立即删除，你自己走还是面试官送你走?如果不是，那过期后到底什么时候被删除呢?﹖是个什么操作?三种不同的删除策略立即删除Redis不可能时时刻刻遍历所有被设置了生存时间的key，来检测数据是否已经到达过期时间，然后对它进行删除。立即删除能保证内存中数据的最大新鲜度，因为它保证过期键值会在过期后马上被删除，其所占用的内存也会随之释放。但是立即删除对cpu是最不友好的。因为删除操作会占用cpu的时间，如果刚好碰上了cpu很忙的时候，比如正在做交集或排序等计算的时候，就会给cpu造成额外的压力，让CPU心累，时时需要删除，忙死。这会产生大量的性能消耗，同时也会影响数据的读取操作。总结:对CPU不友好，用处理器性能换取存储空间(拿时间换空间)惰性删除数据到达过期时间，不做处理。等下次访问该数据时，如果未过期，返回数据﹔发现已过期，删除，返回不存在。惰性删除策略的缺点是，它对内存是最不友好的。如果一个键已经过期，而这个键又仍然保留在redis中，那么只要这个过期键不被删除，它所占用的内存就不会释放。在使用惰性删除策略时，如果数据库中有非常多的过期键，而这些过期键又恰好没有被访问到的话，那么它们也许永远也不会被删除(除非用户手动执行FLUSHDB)，我们甚至可以将这种情况看作是一种内存泄漏–无用的垃圾数据占用了大量的内存，而服务器却不会自己去释放它们，这对于运行状态非常依赖于内存的Redis服务器来说,肯定不是一个好消息总结:对memory不友好，用存储空间换取处理器性能（拿空间换时间)开启惰性淘汰，lazyfree-lazy-eviction=yes以上两种都很极端---定期删除定期删除策略是前两种策略的折中:定期删除策略每隔一段时间执行一次删除过期键操作并通过限制删除操作执行时长和频率来减少删除操作对CPU时间的影响。周期性轮询redis库中的时效性数据，采用随机抽取的策略，利用过期数据占比的方式控制删除频度特点1:CPU性能占用设置有峰值，检测频度可自定义设置特点2:内存压力不是很大，长期占用内存的冷数据会被持续清理总结:周期性抽查存储空间（随机抽查，重点抽查)还会有漏洞1定期删除时，从来没有被抽查到2惰性删除时，也从来没有被点中使用过上述两个步骤======>大量过期的key堆积在内存中，导致redis内存空间紧张或者很快耗尽redis缓存淘汰策略往下看！！！！配置文件redis清理内存的方式?定期删除和惰性删除了解过吗redis缓存淘汰策略有哪些?分别是什么?你用那个?2个维度-----过期键中筛选,所有键中筛选4个方面-----LRU,LFU,random,ttl平时使用在所有的 key都是最近最经常使用，那么就需要选择allkeys-Iru进行置换最近最不经常使用的key，如果你不确定使用哪种策略，那么推荐使用allkeys-lru如果所有的 key的访问概率都是差不多的，那么可以选用allkeys-random 策略去置换数据如果对数据有足够的了解，能够为key 指定 hint(通过expire/ttl指定)，那么可以选择volatile-ttl进行置换如何配置、修改---直接用config命令---直接redis.conf配置文件redis缓存淘汰策略配置性能建议避免仔储bigkey开启惰性淘汰，lazyfree-lazy-eviction=yesredis的LRU了解过吗?请手写LRU----算法手写code见阳哥大厂第3季视频lru和lfu算法的区别是什么区别LRU:最近最少使用页面置换算法，淘汰最长时间未被使用的页面，看页面最后一次被使用到发生调度的时间长短，首先淘汰最长时间未被使用的页面。LFU:最近最不常用页面置换算法，淘汰一定时期内被访问次数最少的页，看一定时间段内页面被使用的频率，淘汰一定时期内被访问次数最少的页举个栗子某次时期Time为10分钟,如果每分钟进行一次调页,主存块为3,若所需页面走向为21212 34假设到页面4时会发生缺页中断若按LRU算法,应换页面1(1页面最久未被使用)，但按LFU算法应换页面3(十分钟内,页面3只使用了一次)可见LRU关键是看页面最后一次被使用到发生调度的时间长短,而LFU关键是看一定时间段内页面被使用的频率!
Redlock算法和底层源码分析当前代码用v8.0的Redis分布式锁-Redlock红锁算法Distributed locks with Redis产生原因----- 单点故障：redis之父提出了Redlock算法解决这个问题Redis也提供了Redlock算法，用来实现基于多个实例的分布式锁。锁变量由多个实例维护，即使有实例发生了故障，锁变量仍然是存在的，客户端还是可以完成锁操作。Redlock算法是实现高可靠分布式锁的一种有效解决方案，可以在实际开发中使用。设计理念该方案为了解决数据不一致的问题，直接舍弃了异步复制只使用master节点，同时由于舍弃了slave，为了保证可用性，引入了N个节点，官方建议是5。演示用3台实例来做说明。客户端只有在满足下面的这两个条件时，才能认为是加锁成功。条件1:客户端从超过半数(大于等于N/2+1）的Redis实例上成功获取到了锁;条件2:客户端获取锁的总耗时没有超过锁的有效时间。redisson实现pom <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.13.4</version> </dependency>RedisConfigpackage com.zzrg.redislock.config; import org.redisson.Redisson; import org.redisson.config.Config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; /** * @Author ZzRG * @Date 2023/6/14 0:39 * @Version 1.0 */ @Configuration public class RedisConfig { @Bean public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); // 配置具体的序列化方式 redisTemplate.setConnectionFactory(lettuceConnectionFactory); // 设置key的序列话方式String redisTemplate.setKeySerializer(new StringRedisSerializer()); // 设置value的序列化方式JSON redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer()); redisTemplate.afterPropertiesSet(); return redisTemplate; } @Bean public Redisson redisson(){ Config config = new Config(); config.useSingleServer().setAddress("redis://192.168.126.130:6379").setDatabase(0).setPassword(""); return (Redisson) Redisson.create(config); } } InventoryServiceImplpackage com.zzrg.redislock.service.Impl; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.config.DistributedLockFactory; import com.zzrg.redislock.myLock.RedisDistributedLock; import com.zzrg.redislock.service.InventoryService; import lombok.extern.slf4j.Slf4j; import org.redisson.Redisson; import org.redisson.api.RLock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.sql.Time; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ @Service @Slf4j public class InventoryServiceImpl implements InventoryService { @Autowired private StringRedisTemplate stringRedisTemplate; @Value("${service.port}") private String port; @Autowired private Redisson redisson; @Override public String saleByRedisson() { String retMessage = ""; RLock redisRedissonLock=redisson.getLock("redisRedissonLock"); redisRedissonLock.lock(); try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage + "\t" + "服务端口号：" + port); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { redisRedissonLock.unlock(); } return retMessage + "\t" + "服务端口号：" + port; } }InventoryServicepackage com.zzrg.redislock.service; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ public interface InventoryService { String sale(); String saleByRedisson(); }InventoryControllerpackage com.zzrg.redislock.controller; import com.zzrg.redislock.service.InventoryService; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; /** * @Author ZzRG * @Date 2023/6/14 1:14 * @Version 1.0 */ @RestController @Api(tags = "redis分布式锁测试") public class InventoryController { @Autowired private InventoryService inventoryService; @ApiOperation("扣减库存sale，一次卖一个") @GetMapping(value = "/inventory/sale") public String sale() { return inventoryService.sale(); } @ApiOperation("扣减库存saleByRedisson，一次卖一个") @GetMapping(value = "/inventory/saleByRedisson") public String saleByRedisson() { return inventoryService.saleByRedisson(); } } 但是还是会出现BUG修改package com.zzrg.redislock.service.Impl; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.config.DistributedLockFactory; import com.zzrg.redislock.myLock.RedisDistributedLock; import com.zzrg.redislock.service.InventoryService; import lombok.extern.slf4j.Slf4j; import org.redisson.Redisson; import org.redisson.api.RLock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.sql.Time; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ @Service @Slf4j public class InventoryServiceImpl implements InventoryService { @Autowired private StringRedisTemplate stringRedisTemplate; @Value("${service.port}") private String port; @Autowired private Redisson redisson; @Override public String saleByRedisson() { String retMessage = ""; RLock redisRedissonLock=redisson.getLock("redisRedissonLock"); redisRedissonLock.lock(); try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage + "\t" + "服务端口号：" + port); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { //改进点，只能删除属于自己的Key，不能删除别人的 if (redisRedissonLock.isLocked() && redisRedissonLock.isHeldByCurrentThread()) { redisRedissonLock.unlock(); } } return retMessage + "\t" + "服务端口号：" + port; } }Redisson源码解析Redis分布式锁过期了，但是业务逻辑还没处理完怎么办缓存续命守护线程"续命"额外起一个线程，定期检查线程是否还持有锁，如果有则延长过期时间。Redisson里面就实现了这个方案，使用“看门狗”定期检查（每1/3的锁时间检查1次），如果线程还持有锁，则刷新过期时间;在获取锁成功后，给锁加一个watchdog, watchdo 会起一个定时任务，在锁没有被释放且快要过期的时候会续期源码分析1通过redisson新建出来的锁key，默认是30秒钟lock 还是juc里面的源码分析2lock() --- tryACquire() ---tryAcquireAsync()源码分析3流程解释通过exists判断，如果锁不存在，则设置值和过期时间，加锁成功通过hexists判断，如果锁已存在，并且锁的是当前线程，则证明是重入锁，加锁成功如果锁已存在，但锁的不是当前线程，则证明有其他线程持有锁。返回当前锁的过期时间(代表了锁key的剩余生存时间)，加锁失败源码分析4解锁
1、初始化版本简单添加2、nginx分布式微服务架构3、redis分布式锁 v3.04、宕机与过期+防止死锁 v4.05、防止误删除key的问题 v5.06、Lua保证原子性 v6.0InventoryControllerpackage com.zzrg.redislock.controller; import com.zzrg.redislock.service.InventoryService; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; /** * @Author ZzRG * @Date 2023/6/14 1:14 * @Version 1.0 */ @RestController @Api(tags = "redis分布式锁测试") public class InventoryController { @Autowired private InventoryService inventoryService; @ApiOperation("扣减库存sale，一次卖一个") @GetMapping(value = "/inventory/sale") public String sale() { return inventoryService.sale(); } }InventoryServicepackage com.zzrg.redislock.service; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ public interface InventoryService { String sale(); }InventoryServiceImpl：package com.zzrg.redislock.service.Impl; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.service.InventoryService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.sql.Time; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ @Service @Slf4j public class InventoryServiceImpl implements InventoryService { @Autowired private StringRedisTemplate stringRedisTemplate; @Value("${service.port}") private String port; /** * v6.0 Lua脚本的redis分布式锁调用，必须保证原子性 * 但是基本 v6.0 版本已经够用 * @return */ @Override public String sale() { String retMessage = ""; String key = "zzrgRedisLock"; String uuidValue = IdUtil.simpleUUID()+":"+Thread.currentThread().getId(); //不用递归了，高并发下容易出错，我们用自旋替代递归方法重试调用；也不用if 用while来代替 //添加Key同时添加过期时间 保证原子性 while(!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue,30L,TimeUnit.SECONDS)){ //等待20毫秒 try { TimeUnit.MILLISECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } //如果这样设置过期时间 就会无法保证原子性 还会出现死锁，添加了key 没走到这一步 就不会有过期时间 //因此需要 与上面的添加Key同时添加过期时间 保证原子性 // stringRedisTemplate.expire(key,30L,TimeUnit.SECONDS); } //枪锁成功的请求线程。进行正常业务逻辑操作，扣减库存 try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { //改进点，修改为Lua脚本的redis分布式锁调用，必须保证原子性，参考官网脚本案例 String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then" + "return redis.call('del',KEYS[1])" + "else " + "return 0 " + "end"; stringRedisTemplate.execute(new DefaultRedisScript(luaScript,Boolean.class), Arrays.asList(key),uuidValue); } return retMessage + "\t" + "服务端口号：" + port; } /** * v4.0 加了过期时间 stringRedisTemplate.opsForValue().setIfAbsent(key, value, 30L, TimeUnit.SECONDS) * 问题：误删锁，如果线程A运行时间超出了过期时间 * 在线程A运行时，xfcyRedisLock这个key过期，另一个线程B进来加了key * 线程A结束后，把线程B的锁删了 * stringRedisTemplate.delete(key); 只能自己删除自己的，需要添加判断是否是自己的锁 * 设置的key + 过期时间不能分开 必须合并在一行具备原子性！ * v5.0 防止误删Key 但是还存在问题是最后的 finally 中的判断+del key 不是一行原子性操作，需要用Lua脚本修改。 * @return @Override public String sale() { String retMessage = ""; String key = "zzrgRedisLock"; String uuidValue = IdUtil.simpleUUID()+":"+Thread.currentThread().getId(); //不用递归了，高并发下容易出错，我们用自旋替代递归方法重试调用；也不用if 用while来代替 //添加Key同时添加过期时间 保证原子性 while(!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue,30L,TimeUnit.SECONDS)){ //等待20毫秒 try { TimeUnit.MILLISECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } //如果这样设置过期时间 就会无法保证原子性 还会出现死锁，添加了key 没走到这一步 就不会有过期时间 //因此需要 与上面的添加Key同时添加过期时间 保证原子性 // stringRedisTemplate.expire(key,30L,TimeUnit.SECONDS); } //枪锁成功的请求线程。进行正常业务逻辑操作，扣减库存 try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { //改进点，只能删除数据自己的key，不能删除被人的 if (stringRedisTemplate.opsForValue().get(key).equalsIgnoreCase(uuidValue)){ stringRedisTemplate.delete(key); } } return retMessage + "\t" + "服务端口号：" + port; } */ /** * v3.2 出现的问题死锁：分布式微服务中突然一个挂了，代码层面跟们没有走到finally，这个服务就一直拿着这个锁，没办法保证解锁(redis 中的key一直存在) * 这个key没有被删除，需要加入一个过期时间的key。 * --------------------------------------------- * setnx 用while判断 * 问题： setnx过后，正在进行业务逻辑操作时，没有走到finally之前，整个微服务down机了，导致锁一直存在 * 不是程序出了问题，如果程序问题，最后还是会执行finally * 没办法保证解锁（没有过期时间，该key一直存在），需要加入过期时间限定key * @return @Override public String sale() { String retMessage = ""; String key = "zzrgRedisLock"; String uuidValue = IdUtil.simpleUUID()+":"+Thread.currentThread().getId(); //不用递归了，高并发下容易出错，我们用自旋替代递归方法重试调用；也不用if 用while来代替 while(!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue)){ //等待20毫秒 try { TimeUnit.MILLISECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } } //枪锁成功的请求线程。进行正常业务逻辑操作，扣减库存 try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { stringRedisTemplate.delete(key); } return retMessage + "\t" + "服务端口号：" + port; } */ /** * v3.1 版本的出现了递归重试，容易导致stackoverflowerror,所以不太推荐; 另外，高并发缓刑后推荐while判断而不是if @Override public String sale() { String retMessage = ""; String key = "zzrgRedisLock"; String uuidValue = IdUtil.simpleUUID()+":"+Thread.currentThread().getId(); Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue); //flag == false 抢不到的线程要继续重试。。。。 if (!flag){ //等待20毫秒，进行递归重试。。。 try { TimeUnit.MILLISECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } sale(); }else { //枪锁成功的请求线程。进行正常业务逻辑操作，扣减库存 try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { stringRedisTemplate.delete(key); } } return retMessage + "\t" + "服务端口号：" + port; } */ /** * v2.0， 单机版枷锁配合nigx和jmeter压测后，不满足高并发分布式锁的性能要求，出现超卖 private Lock lock = new ReentrantLock(); @Override public String sale() { String retMessage = ""; lock.lock(); try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { lock.unlock(); } return retMessage + "\t" + "服务端口号：" + port; } */ @Override public String saleByRedisson() { } }//改进点，修改为Lua脚本的redis分布式锁调用，必须保证原子性，参考官网脚本案例 String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then" + "return redis.call('del',KEYS[1])" + "else " + "return 0 " + "end"; stringRedisTemplate.execute(new DefaultRedisScript(luaScript,Boolean.class), Arrays.asList(key),uuidValue);7、可重入锁+设计模式可重入锁又名递归锁是指在同一个线程在外层方法获取锁的时候，再进入该线程的内层方法会自动获取锁(前提，锁对象得是同一个对象)，不会因为之前已经获取过还没释放而阻塞。如果是1个有synchronized修饰的递归调用方法，程序第2次进入被自己阻塞了岂不是天大的笑话，出现了作茧自缚。所以Java中ReentrantLock和synchronized都是可重入锁，可重入锁的一个优点是可一定程度避免死锁。隐式锁（即synchronized关键字使用的锁)默认是可重入锁指的是可重复可递归调用的锁，在外层使用锁之后，在内层仍然可以使用，并且不发生死锁，这样的锁就叫做可重入锁。简单的来说就是:在一个synchronized修饰的方法或代码块的内部调用本类的其他synchronized修饰的方法或代码块时，是永远可以得到锁的与可重入锁相反，不可重入锁不可递归调用，递归调用就发生死锁。每个锁对象拥有一个锁计数器和一个指向持有该锁的线程的指针。当执行monitorenter时，如果目标锁对象的计数器为零，那么说明它没有被其他线程所持有，Java虚拟机会将该锁对象的持有线程设置为当前线程，并且将其计数器加1。在目标锁对象的计数器不为零的情况下，如果锁对象的持有线程是当前线程，那么]ava虚拟机可以将其计数器加1，否则需要等待，直至持有线程释放该锁。当执行monitorexit时，Java虚拟机则需将锁对象的计数器减1。计数器为零代表锁已被释放。//加锁的Lua脚本，对标我们的lock方法 v1.0 if redis.call('exists','key') ==0 then redis.call('hest','key','uuid:threadid',1) redis.call('expire','key',50) return 1 elseif redis.call('hexists','key','uuid:threadid') ==1 then redis.call('hincrby','key','uuid:threadid',1) redis.call('expire','key',50) return 1 else return 0 end //v2.0 合并相同的代码，用hincrby替代hset，精简代码 if redis.call('exists','key') ==0 or redis.call('hexists','key','uuid:threadid') == 1 then redis.call('hincrby','key','uuid:threadid',1) redis.call('expire','key',50) return 1 else return 0 end //v3.0 脚本换成动态参数来代替 if redis.call('exists',KEYS[1]) ==0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then redis.call('hincrby',KEYS[1],ARGV[1],1) redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end //解锁的Lua脚本，对标我们的unlock方法//V1 if redis.call( 'hexists ',key , uuid:threadid) == 0 then return nil elseif redis.call( 'hincrby' , key, uuid:threadid,-1)== 0 then return redis.call( 'del ' ,key) else return 0 end //V2 if redis.call( ' HEXISTS',KEYS[1],ARGV[1])== 0 then return nil elseif redis.cal1( 'HINCRBY' ,KEYS[1],ARGV[1],-1) == 0 then return redis.call( 'del',KEYS[1]) else return 0 end 程序编写创建RedisDistributedLockpackage com.zzrg.redislock.myLock; import cn.hutool.core.util.IdUtil; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; /** * @Author ZzRG * @Date 2023/6/17 16:42 * @Version 1.0 * 我们自研的redis分布式锁，实现了Lock接口 */ public class RedisDistributedLock implements Lock { private StringRedisTemplate stringRedisTemplate; private String lockName; //KEYS[1] private String uuidValue; //ARGV[1] private long expireTime; //ARGV[2] public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName) { this.stringRedisTemplate = stringRedisTemplate; this.lockName = lockName; this.uuidValue = IdUtil.simpleUUID()+":" + Thread.currentThread().getId(); this.expireTime = 50L; } @Override public void lock() { tryLock(); } @Override public boolean tryLock() { try { tryLock(-1L,TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } return false; } @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (time == -1L){ String script = "if redis.call('exists',KEYS[1]) ==0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then" + "redis.call('hincrby',KEYS[1],ARGV[1],1)" + "redis.call('expire',KEYS[1],ARGV[2])" + "return 1" + "else" + "return 0" + "end "; while (!stringRedisTemplate.execute(new DefaultRedisScript<>(script,Boolean.class), Arrays.asList(lockName),uuidValue,String.valueOf(expireTime))){ //等待20毫秒 try { TimeUnit.MILLISECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } } return true; } return false; } @Override public void unlock() { String script = "if redis.call( ' HEXISTS',KEYS[1],ARGV[1])== 0 then" + "return nil" + "elseif redis.cal1( 'HINCRBY' ,KEYS[1],ARGV[1],-1) == 0 then" + "return redis.call( 'del',KEYS[1])" + "else" + "return 0" + "end"; // nil =false 1=true 0=false Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime)); if (null ==flag){ throw new RuntimeException("this lock doesn't exists"); } } // ===========下面两个暂时用不到 不在重写！===== @Override public void lockInterruptibly() throws InterruptedException { } @Override public Condition newCondition() { return null; } } 业务逻辑package com.zzrg.redislock.service.Impl; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.myLock.RedisDistributedLock; import com.zzrg.redislock.service.InventoryService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.sql.Time; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ @Service @Slf4j public class InventoryServiceImpl implements InventoryService { @Autowired private StringRedisTemplate stringRedisTemplate; @Value("${service.port}") private String port; //v7.0 如何将我们的Lock/unLock+lua脚本自研版的redis分布式锁搞定?| Lock redisDistributedLock=new RedisDistributedLock(stringRedisTemplate,"RedisLock"); @Override public String sale() { String retMessage = ""; redisDistributedLock.lock(); try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { redisDistributedLock.unlock(); } return retMessage + "\t" + "服务端口号：" + port; } }这样设计出现了分布式锁写死的问题改造引入工厂模式package com.zzrg.redislock.config; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.myLock.RedisDistributedLock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import java.util.concurrent.locks.Lock; /** * @Author ZzRG * @Date 2023/6/17 17:23 * @Version 1.0 */ @Component public class DistributedLockFactory { @Autowired private StringRedisTemplate stringRedisTemplate; private String lockName; private String uuidValue; public Lock getDistributedLock(String lockType) { if (lockType == null) { return null; } if (lockType.equalsIgnoreCase("REDIS")) { this.lockName = "RedisLock"; return new RedisDistributedLock(stringRedisTemplate,lockName); } else if (lockType.equalsIgnoreCase("ZOOKEEPER")) { this.lockName = "ZookeeperLock"; // TODO zoookeeper 版本的分布式锁 return null; } else if (lockType.equalsIgnoreCase("MYSQL")) { this.lockName = "MysqlLock"; // TODO MYSQL 版本的分布式锁 return null; } return null; } } 引入可重入锁业务逻辑package com.zzrg.redislock.service.Impl; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.config.DistributedLockFactory; import com.zzrg.redislock.myLock.RedisDistributedLock; import com.zzrg.redislock.service.InventoryService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.sql.Time; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ @Service @Slf4j public class InventoryServiceImpl implements InventoryService { @Autowired private StringRedisTemplate stringRedisTemplate; @Value("${service.port}") private String port; @Autowired private DistributedLockFactory distributedLockFactory; //v7.0 如何将我们的Lock/unLock+lua脚本自研版的redis分布式锁搞定? @Override public String sale() { String retMessage = ""; Lock redisLock=distributedLockFactory.getDistributedLock("REDIS"); redisLock.lock(); try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage + "\t" + "服务端口号：" + port); testReEntry(); } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { redisLock.unlock(); } return retMessage + "\t" + "服务端口号：" + port; } //可重性测试 private void testReEntry() { Lock redisLock=distributedLockFactory.getDistributedLock("REDIS"); redisLock.lock(); try { System.out.println("==========测试可重入锁=========="); } finally { redisLock.unlock(); } }中间加入可重入性测试出现问题改造DistributedLockFactorypackage com.zzrg.redislock.config; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.myLock.RedisDistributedLock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import java.util.concurrent.locks.Lock; /** * @Author ZzRG * @Date 2023/6/17 17:23 * @Version 1.0 */ @Component public class DistributedLockFactory { @Autowired private StringRedisTemplate stringRedisTemplate; private String lockName; private String uuidValue; public DistributedLockFactory() { this.uuidValue = IdUtil.simpleUUID(); } public Lock getDistributedLock(String lockType) { if (lockType == null) { return null; } if (lockType.equalsIgnoreCase("REDIS")) { this.lockName = "RedisLock"; return new RedisDistributedLock(stringRedisTemplate,lockName,uuidValue); } else if (lockType.equalsIgnoreCase("ZOOKEEPER")) { this.lockName = "ZookeeperLock"; // TODO zoookeeper 版本的分布式锁 return null; } else if (lockType.equalsIgnoreCase("MYSQL")) { this.lockName = "MysqlLock"; // TODO MYSQL 版本的分布式锁 return null; } return null; } } 改造RedisDistributedLock // public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName) { // this.stringRedisTemplate = stringRedisTemplate; // this.lockName = lockName; // this.uuidValue = IdUtil.simpleUUID()+":" + Thread.currentThread().getId(); // this.expireTime = 50L; // } public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName,String uuidValue) { this.stringRedisTemplate = stringRedisTemplate; this.lockName = lockName; this.uuidValue = uuidValue+":" + Thread.currentThread().getId(); this.expireTime = 50L; }8、自动续期//自动续期加个钟的Lua脚本 if redis.call('HEISTS'，KEYS[1],ARGV[1]) ==1 then return redis.call('expire',KEYS[1],ARGV[2]) else return 0 endpackage com.zzrg.redislock.myLock; import cn.hutool.core.util.IdUtil; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import java.util.Arrays; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; /** * @Author ZzRG * @Date 2023/6/17 16:42 * @Version 1.0 * 我们自研的redis分布式锁，实现了Lock接口 */ //@component引入DistributedLockFactory 工厂模式，从工厂获得即可 public class RedisDistributedLock implements Lock { private StringRedisTemplate stringRedisTemplate; private String lockName; //KEYS[1] private String uuidValue; //ARGV[1] private long expireTime; //ARGV[2] // public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName) { // this.stringRedisTemplate = stringRedisTemplate; // this.lockName = lockName; // this.uuidValue = IdUtil.simpleUUID()+":" + Thread.currentThread().getId(); // this.expireTime = 50L; // } public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName,String uuidValue) { this.stringRedisTemplate = stringRedisTemplate; this.lockName = lockName; this.uuidValue = uuidValue+":" + Thread.currentThread().getId(); this.expireTime = 50L; } @Override public void lock() { tryLock(); } @Override public boolean tryLock() { try { tryLock(-1L,TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } return false; } @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (time == -1L){ String script = "if redis.call('exists',KEYS[1]) ==0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then" + "redis.call('hincrby',KEYS[1],ARGV[1],1)" + "redis.call('expire',KEYS[1],ARGV[2])" + "return 1" + "else" + "return 0" + "end "; System.out.println("lockName:" + lockName + "\t" + "uuidValue" + uuidValue); while (!stringRedisTemplate.execute(new DefaultRedisScript<>(script,Boolean.class), Arrays.asList(lockName),uuidValue,String.valueOf(expireTime))){ //等待20毫秒 try { TimeUnit.MILLISECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } } //新建一个后台扫描程序，来坚持key目前的ttl，是否到我们规定的1/2 1/3未来实现续期 this.renewExpire(); return true; } return false; } @Override public void unlock() { String script = "if redis.call( ' HEXISTS',KEYS[1],ARGV[1])== 0 then" + "return nil" + "elseif redis.cal1( 'HINCRBY' ,KEYS[1],ARGV[1],-1) == 0 then" + "return redis.call( 'del',KEYS[1])" + "else" + "return 0" + "end"; // nil =false 1=true 0=false Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime)); if (null ==flag){ throw new RuntimeException("this lock doesn't exists"); } } private void renewExpire() { String script = "if redis.call('HEISTS'，KEYS[1],ARGV[1]) ==1 then" + "return redis.call('expire',KEYS[1],ARGV[2])" + "else" + "return 0" + "end"; new Timer().schedule(new TimerTask() { @Override public void run() { if (stringRedisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime))) { renewExpire(); } } },(this.expireTime * 1000/3)); } // ===========下面两个暂时用不到 不在重写！===== @Override public void lockInterruptibly() throws InterruptedException { } @Override public Condition newCondition() { return null; } } package com.zzrg.redislock.service.Impl; import cn.hutool.core.util.IdUtil; import com.zzrg.redislock.config.DistributedLockFactory; import com.zzrg.redislock.myLock.RedisDistributedLock; import com.zzrg.redislock.service.InventoryService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import java.sql.Time; import java.util.Arrays; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @Author ZzRG * @Date 2023/6/14 1:09 * @Version 1.0 */ @Service @Slf4j public class InventoryServiceImpl implements InventoryService { @Autowired private StringRedisTemplate stringRedisTemplate; @Value("${service.port}") private String port; @Autowired private DistributedLockFactory distributedLockFactory; //V8.0，实现自动续期功能的完善，后台自定义扫描程序，如果规定时间内没有完成业务逻辑，会调用加钟自动续期的脚本nublic string ca1e( ) @Override public String sale() { String retMessage = ""; Lock redisLock=distributedLockFactory.getDistributedLock("REDIS"); redisLock.lock(); try { // 1 查询库存信息 String result = stringRedisTemplate.opsForValue().get("inventory001"); // 2 判断库存是否足够 Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result); // 3 扣减库存 每次减少一个 if (inventoryNumber > 0) { stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber)); retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber; System.out.println(retMessage + "\t" + "服务端口号：" + port); //暂停120秒钟线程 故意的 主要演示自动续期的功能 try { TimeUnit.MILLISECONDS.sleep(120); } catch (InterruptedException e) { e.printStackTrace(); } } else { retMessage = "商品卖完了，o(╥﹏╥)o"; } } finally { redisLock.unlock(); } return retMessage + "\t" + "服务端口号：" + port; } }
面试题Redis除了拿来做缓存，你还见过基于Redis的什么用法?数据共享，分布式Session分布式锁全局ID计算器、点赞位统计购物车轻量级消息队列---list、stream抽奖点赞、签到、打卡差集交集并集，用户关注、可能认识的人，推荐模型热点新闻、热搜排行榜Redis 做分布式锁的时候有需要注意的问题?你们公司自己实现的分布式锁是否用的setnx命令实现(大厂一定不可以)?这个是最合适的吗?你如何考虑分布式锁的可重入问题?如果是Redis是单点部署的，会带来什么问题?Redis集群模式下，比如主从模式，CAP方面有没有什么问题呢?那你简单的介绍一下Redlock吧?你简历上写redisson，你谈谈Redis分布式锁如何续期?看门狗知道吗?锁的种类单机版同一个JVM虚拟机内，synchronized或者Lock接口分布式多个不同JⅣM虚拟机，单机的线程锁机制不再起作用，资源类在不同的服务器之间共享了。一个靠谱分布式锁需要具备的条件和刚需独占性----OnlyOne，任何时刻只能有且仅有一个线程持有高可用----若redis集群环境下，不能因为某一个节点挂了而出现获取锁和释放锁失败的情况----高并发请求下，依l旧性能OK好使防死锁----杜绝死锁，必须有超时控制机制或者撤销操作，有个兜底终止跳出方案不乱抢----防止张冠李戴，不能私下unlock别人的锁，只能自己加锁自己释放，自己约的锁含着泪也要自己解重入性----同一个节点的同一个线程如果获得锁之后，它也可以再次获取这个锁。分布式锁setnx key valueset key value [EX seconds] [Px milliseconds] [NX|XX]JUC中A锁的规范落地参考+可重入锁考虑+Lua脚本+Redis命令一步步实现分布式锁Lua脚本Lua是—种轻量小巧的脚本语言用标准C语言编写并以源代码形式开放，其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。Lua是巴西里约热内卢天主教大学(Pontifical Catholic University of Rio de Janeiro)里的一个研究小组于1993年开发的，该小组成员有:Roberto lerusalimschy、waldemar Celes和Luiz Henrique de Figueiredo。设计目的：其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。Lua特性轻量级:它用标准C语言编写并以源代码形式开放，编译后仅仅一百余K，可以很方便的嵌入别的程序里。可扩展: Lua提供了非常易于使用的扩展接口和机制:由宿主语言(通常是C或C++)提供这些功能，Lua可以使用它们，就像是本来就内置的功能一样。Redis调用Lua脚本通过eval命令保证代码执行的原子性，直接用return返回脚本执行后的结果值入门级：reids中 eval luascript numkeys [key [key ...]] [arg [arg ...]] EVAL "return 'hello lua'" 0 ----------------------------------------------------------- set k1 v1 expire k1 30 get k1 EVAL "redis.call('set','k1','v1') redis.call('expire','k1','30') return redis.call('get','k1')" 0 ----------------------------------------------------------- mset k1 v1 k2 v2 EVAL "redis.call('mset','k1','v1','k2','v2')" 0 //动态的 上面案例 后面的0代表 没有传任何参数 EVAL "redis.call('mset',KEYS[1],ARGV[1],KEYS[2],ARGV[2])" 2 k1 k2 v1 v2升级 这个没用到redis中 条件判断：if(布尔条件) then 业务代码 elseif(布尔条件) then 业务代码 else 业务代码 end ------------------------------------------------------------ EVAL "if KEYS[2]>KEYS[2] then return ARGV[1] elseif KEYS[1]<KEYS[2] then return ARGV[2] else ARGV[3] end" 2 5 7 1 2 3 // if(5>7){ return 1 }else if(5<7){ return 2 }else{ return 3 } //结果"2"