坏代码导致的性能问题大赏：CPU占用飙到了900%！ FGC

发表于 2021-10-31

读过《重构 - 改善既有代码的设计》一书的同学们应该都很了解“代码的坏味道”。当然确定什么是代码“坏味道”是主观的，它会随语言、开发人员和开发方法的不同而不同。在工作当中，很多时候都是在维护之前的项目和在此基础上增加一些新功能，为了能让项目代码易于理解和维护，要时刻注意代码中的“坏味道”，当发现代码如果有坏味道了，要及时去重构它使其变成优秀的整洁的代码。今天我们要聊的是“坏味道的代码”给系统性能带来的影响，笔者会给大家展示几个案例，希望能对大家有所启发和帮助。

FGC实战：坏代码导致服务频繁FGC无响应问题分析

问题

网络问题？

晚上七点多开始，我就开始不停地收到报警邮件，邮件显示探测的几个接口有超时情况。多数执行栈都在：

arduino复制代码java.io.BufferedReader.readLine(BufferedReader.java:371)
java.io.BufferedReader.readLine(BufferReader.java:389)
java_io_BufferedReader$readLine.call(Unknown Source)
com.domain.detect.http.HttpClient.getResponse(HttpClient.groovy:122)
com.domain.detect.http.HttpClient.this$2$getResponse(HttpClient.groovy)

这个线程栈的报错我见得多了，我们设置的 HTTP DNS 超时是 1s， connect 超时是 2s， read 超时是 3s，这种报错都是探测服务正常发送了 HTTP 请求，服务器也在收到请求正常处理后正常响应了，但数据包在网络层层转发中丢失了，所以请求线程的执行栈会停留在获取接口响应的地方。这种情况的典型特征就是能在服务器上查找到对应的日志记录。而且日志会显示服务器响应完全正常。与它相对的还有线程栈停留在 Socket connect 处的，这是在建连时就失败了，服务端完全无感知。

我注意到其中一个接口报错更频繁一些，这个接口需要上传一个 4M 的文件到服务器，然后经过一连串的业务逻辑处理，再返回 2M 的文本数据，而其他的接口则是简单的业务逻辑，我猜测可能是需要上传下载的数据太多，所以超时导致丢包的概率也更大吧。

根据这个猜想，群登上服务器，使用请求的 request_id 在近期服务日志中搜索一下，果不其然，就是网络丢包问题导致的接口超时了。

当然这样 leader 是不会满意的，这个结论还得有人接锅才行。于是赶紧联系运维和网络组，向他们确认一下当时的网络状态。网络组同学回复说是我们探测服务所在机房的交换机老旧，存在未知的转发瓶颈，正在优化，这让我更放心了，于是在部门群里简单交待一下，算是完成任务。

问题爆发

本以为这次值班就起这么一个小波浪，结果在晚上八点多，各种接口的报警邮件蜂拥而至，打得准备收拾东西过周日单休的我措手不及。

这次几乎所有的接口都在超时，而我们那个大量网络 I/O 的接口则是每次探测必超时，难道是整个机房故障了么。

我再次通过服务器和监控看到各个接口的指标都很正常，自己测试了下接口也完全 OK，既然不影响线上服务，我准备先通过探测服务的接口把探测任务停掉再慢慢排查。

结果给暂停探测任务的接口发请求好久也没有响应，这时候我才知道没这么简单。

解决

内存泄漏
于是赶快登陆探测服务器，首先是 top free df 三连，结果还真发现了些异常。

我们的探测进程 CPU 占用率特别高，达到了 900%。

我们的 Java 进程，并不做大量 CPU 运算，正常情况下，CPU 应该在 100~200% 之间，出现这种 CPU 飙升的情况，要么走到了死循环，要么就是在做大量的 GC。

使用 jstat -gc pid [interval] 命令查看了 java 进程的 GC 状态，果然，FULL GC 达到了每秒一次。

这么多的 FULL GC，应该是内存泄漏没跑了，于是使用 jstack pid > jstack.log 保存了线程栈的现场，使用 jmap -dump:format=b,file=heap.log pid 保存了堆现场，然后重启了探测服务，报警邮件终于停止了。

jstat

jstat 是一个非常强大的 JVM 监控工具，一般用法是： jstat [-options] pid interval

它支持的查看项有：

class 查看类加载信息
compile 编译统计信息
gc 垃圾回收信息
gcXXX 各区域 GC 的详细信息如 -gcold

使用它，对定位 JVM 的内存问题很有帮助。

排查

问题虽然解决了，但为了防止它再次发生，还是要把根源揪出来。

分析栈
栈的分析很简单，看一下线程数是不是过多，多数栈都在干嘛。

1 2	shell复制代码> grep 'java.lang.Thread.State' jstack.log \| wc -l > 464

才四百多线程，并无异常。

bash复制代码> grep -A 1 'java.lang.Thread.State' jstack.log  | grep -v 'java.lang.Thread.State' | sort | uniq -c |sort -n

     10  at java.lang.Class.forName0(Native Method)
     10  at java.lang.Object.wait(Native Method)
     16  at java.lang.ClassLoader.loadClass(ClassLoader.java:404)
     44  at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
    344  at sun.misc.Unsafe.park(Native Method)

线程状态好像也无异常，接下来分析堆文件。

下载堆 dump 文件
堆文件都是一些二进制数据，在命令行查看非常麻烦，Java 为我们提供的工具都是可视化的，Linux 服务器上又没法查看，那么首先要把文件下载到本地。

由于我们设置的堆内存为 4G，所以 dump 出来的堆文件也很大，下载它确实非常费事，不过我们可以先对它进行一次压缩。

gzip 是个功能很强大的压缩命令，特别是我们可以设置 -1 ~ -9 来指定它的压缩级别，数据越大压缩比率越大，耗时也就越长，推荐使用 -6~7， -9 实在是太慢了，且收益不大，有这个压缩的时间，多出来的文件也下载好了。

使用 MAT 分析 jvm heap
MAT 是分析 Java 堆内存的利器，使用它打开我们的堆文件（将文件后缀改为 .hprof）, 它会提示我们要分析的种类，对于这次分析，果断选择 memory leak suspect。

从上面的饼图中可以看出，绝大多数堆内存都被同一个内存占用了，再查看堆内存详情，向上层追溯，很快就发现了罪魁祸首。

分析代码
找到内存泄漏的对象了，在项目里全局搜索对象名，它是一个 Bean 对象，然后定位到它的一个类型为 Map 的属性。

这个 Map 根据类型用 ArrayList 存储了每次探测接口响应的结果，每次探测完都塞到 ArrayList 里去分析，由于 Bean 对象不会被回收，这个属性又没有清除逻辑，所以在服务十来天没有上线重启的情况下，这个 Map 越来越大，直至将内存占满。

内存满了之后，无法再给 HTTP 响应结果分配内存了，所以一直卡在 readLine 那。而我们那个大量 I/O 的接口报警次数特别多，估计跟响应太大需要更多内存有关。

给代码 owner 提了 PR，问题圆满解决。

小结

其实还是要反省一下自己的，一开始报警邮件里还有这样的线程栈：

csharp复制代码groovy.json.internal.JsonParserCharArray.decodeValueInternal(JsonParserCharArray.java:166)
groovy.json.internal.JsonParserCharArray.decodeJsonObject(JsonParserCharArray.java:132)
groovy.json.internal.JsonParserCharArray.decodeValueInternal(JsonParserCharArray.java:186)
groovy.json.internal.JsonParserCharArray.decodeJsonObject(JsonParserCharArray.java:132)
groovy.json.internal.JsonParserCharArray.decodeValueInternal(JsonParserCharArray.java:186)

看到这种报错线程栈却没有细想，要知道 TCP 是能保证消息完整性的，况且消息没有接收完也不会把值赋给变量，这种很明显的是内部错误，如果留意后细查是能提前查出问题所在的，查问题真是差了哪一环都不行啊。

来源 | zhenbianshu.github.io/

记一次jvm疯狂gc导致CPU飙高的问题解决

背景

线上web服务器不时的出现非常卡的情况，登录服务器top命令发现服务器CPU非常的高，重启tomcat之后CPU恢复正常，半天或者一天之后又会偶现同样的问题。解决问题首先要找到问题的爆发点，对于偶现的问题是非常难于定位的。

重启服务器之后只能等待问题再次出现，这时候首先怀疑是否某个定时任务引发大量计算或者某个请求引发了死循环，所以先把代码里面所有怀疑的地方分析了一遍，加了一点日志，结果第二天下午问题再次出现，

这次的策略是首先保护案发现场，因为线上是两个点，把一个点重启恢复之后把另一个点只下线不重启保留犯罪现场。

排查

在问题的服务器上首先看业务日志，没有发现大量重复日志，初步排除死循环的可能，接下来只能分析jvm了。

第一步：top命令查看占用CPU的pid

这是事后的截图，当时的cpu飙高到500多，pid是27683

然后ps aux | grep 27683 搜索一下确认一下是不是我们的tomcat占用的cpu，这个基本是可以肯定的，因为tomcat重启之后CPU立马就降下来了。

也可以使用jps显示java的pid

第二步：top -H -p 27683 查找27683下面的线程id，显示线程的cpu的占用时间，占用比例，发现有很多个线程都会CPU占用很高，只能每个排查。

第三步：jstack查看线程信息，命令：jstack 27683 >> aaaa.txt 输出到文本中再搜索，也可以直接管道搜索 jstack 27683 | grep "6c23" 这个线程id是16进制表示，需要转一下，可以用这个命令转 printf "%x\n" tid 也可以自己计算器转一下。

悲剧的是我在排查的时候被引入了一个误区，当时搜索到6c26这个线程的时候，发现是在做gc，疯狂gc导致的线程过高，但是找不到哪里造成的产生这么多对象，一直在找所有可能的死循环和可能的内存泄露。

然后通过命名 jstat -gcutil 【PID】 1000 100 查看每秒钟gc的情况。

这个是事后复盘的截图，当时的截图已经没有了

发现S0不停的再新建对象，然后gc，不停的反复如此gc，去看堆栈信息，没有什么发现，无非都是String和map对象最多，确定不了死循环的代码，也找不到问题的爆发点，至此陷入彻底的困惑。一番查找之后确认也不是内存泄露，苦苦寻找无果，我陷入了沉思。

CPU还是一直保持在超高，无奈之下，还是jstack 27683 看线程栈，无目的的乱看，但是发现了一个问题，当前的点我是下线的也就是没有用户访问的，CPU还是一直这么高，而且线程栈也在不停的打印，那么也就是说当前还在运行的线程很可能就是元凶，马上分析线程栈信息，有了重大发现。

大量的这个线程信息出现，httpProxy_jsp这个jsp的线程在不停的活跃，这个是什么鬼jsp？难道服务器被攻击了？马上去代码里找，发现确实有这个jsp，查看git的提交记录，是几天之前另一个同事提交的代码，时间点和问题第一次出现的时间非常吻合，感觉自己运气好应该是找到问题的点了，把同事叫过来分析他的代码，这个jsp其实非常简单，就是做一个代理请求，发起了一个后端http请求。

HttpRequestUtil如下，是同事自己写的工具类，没有用公用工具，其中一个post方法里connection没有设置链接超时时间和read超时时间：

这里面有个致命的问题，他http请求没有设置超时等待时间，connection如果不设置超时时间或者0就认为是无穷大，也就是会一直都不超时，这时候如果被请求的第三方页面如果不响应或者响应非常慢，这个请求就会一直的等待，或者是请求没回来接着又来一次，导致这个线程就卡住了，但是线程堆积在这里又不崩溃还一直的在做某些事情会产生大量的对象，然后触发了jvm不停的疯狂GC把服务器CPU飚到了极限，然后服务器响应变得非常慢，问题终于找到了，而且非常符合问题的特点，果然把这个地方换了一种写法加了2秒钟超时的限制，问题没有再出现。

这次问题的解决过程得出几点心得：

1、jvm的问题是很复杂的，通过日志看到的很可能不是问题的根源，解决问题也有运气成分，分析日志+业务场景+瞎蒙都是解决问题的方法，分析问题不要一条道走到黑，多结合当前的场景加上一些猜测。

2、这个问题的根源是CPU飙高，一开始总想着是代码里有死循环，后来又以为是内存泄露，所以走了很多弯路，最后还是用最笨的方法看线程栈的日志看出了问题，所以问题没有统一的解决方案，具体问题都要具体处理的，不能拘泥于以往的经验。

3、在写代码过程中尽量使用原项目中已经被广泛使用的公共工具类，尽量不要把自己自创的没有经过项目检验的代码引入工程，即使看起来很简单的一段代码可能给项目引入灾难，除非你有充足的把握了解你代码的底层，比如这个超时的设置问题。

记一次Synchronized关键字使用不合理，导致的多线程下线程阻塞问题排查

在为客户进行性能诊断调优时，碰到了一个Synchronized关键字使用不合理导致多线程下线程阻塞的情况。用文字记录下了问题的整个发现-排查-分析-优化过程，排查过程中使用了我司商业化产品——XLand性能分析平台，通过文章主要希望跟大家分享下分析和优化思路以及注意点，有兴趣深入了解的同学可以评论交流。

现象

在执行单接口负载“判断登陆是否正常接口”测试时候，发现10用户增加至50用户并发，TPS保持不变，响应时间处于持续递增状态，应用CPU为27%，数据库CPU为3%，资源消耗维持稳定状态，由此判断应用程序可能存在瓶颈。

分析

通过XLand分析平台线程分析，发现某线程存在锁等待情况，通过XLand中的x分析定位，发现AuthProvider类中getAccessor方法有Synchronized关键字，当两个以上线程同时调用该同步方法时，每次只能有一个线程能进入该方法，其他线程必须等前一个线程执行完该同步方法后，才能有机会进入。
马赛克4.png

风险点

Synchronized关键字解决的是多个线程之间访问资源的同步性，Synchronized关键字可以保证被它修饰的方法或者代码块在任意时刻只能有一个线程执行。谨慎使用Synchronized关键字，以防导致不必要的线程阻塞，影响响应时间。

优化措施

把AuthProvider类中的Synchronized关键字去掉，发现在10用户并发下判断登陆是否正常接口TPS由原来的174笔/秒增长至624笔/秒，增长了3倍。在日常编程中谨慎使用synchronized，如果没有多线程修改静态变量或单例属性这类需求就不要用，如果有需要也建议只锁必要的代码块，而不是锁整个方法。

后记

Java 应用性能的瓶颈点非常多，比如磁盘、内存、网络 I/O 等系统因素，Java 应用代码，JVM GC，数据库，缓存等。一般将 Java 性能优化分为 4 个层级：应用层、数据库层、框架层、JVM 层。每层优化难度逐级增加，涉及的知识和解决的问题也会不同。但说实话，其实大多数问题还没有需要你懂框架源代码、JVM参数、GC工作机制这一步，只需要略会分析SQL，理解代码逻辑，会定位到有问题的Java代码并作修改即可。毕竟不是有这么一句话是这么说来着——80%的性能问题都是你写的烂代码导致的，哈哈哈。虽然有点犀利，但是保持良好的编码习惯，合理使用某些可能引起问题的关键字，谨慎使用内存资源，的确能规避很大一部分问题。好了，最后祝大家都徒手千行无bug！

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

final关键字你真的会了吗什么是final final常

发表于 2021-10-31

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动。

什么是final

final是java中的一个关键字，可以修饰变量（成员变量+局部变量）、方法以及类。

final常量

final修饰常量有以下特点：

如果引用为基本数据类型，则该引用为常量，该值无法修改；

如果引用为引用数据类型，比如对象、数组，则该对象、数组本身可以修改，但指向该对象或数组的地址的引用不能修改。

如果引用时类的成员变量，则必须当场赋值，否则编译会报错：定义时赋值或构造函数内赋值

final修饰常量基本类型和引用类型有些不同。

基本数据类型：

arduino复制代码 private final int A;
 private final int B = 3;
 private final int C = new Random().nextInt();
 public Test() {
 this.A = 5;
 }
 public Test(int a) {
 this.A = a;
 }
}

像B这种确定值(定义便立即赋值)的final常量，编译器会在编译时将该常量值带入到任何可能用到它的计算式中，这会减轻运行时的一些负担。

引用类型：

引用类型不能改变指的是引用被初始化指向一个对象后，就再也无法改为指向其他对象，但其指向的对象本身是可以被修改的。

typescript复制代码 //final常量d指向了对象D
 private final D d =new D();
 //这个方法改变了对象D的内容
 public D change(){
 this.d.setS("change");
 return d;
 }
 //注意这个方法,将常量d的引用改变了
 public D refnewD(){
 this.d = new D();
 return d;
 }
 class D {
 private String s= "init";
 public String getS() {
 return s;
 }
 public void setS(String s) {
 this.s = s;
 }
 }
}

上面代码是无法编译的，因为this.d=new D()会提示Cannot assign a value to final variable ‘d’,但是你却可以调用change()方法来改变d对象的s变量值。

final修饰方法

final修饰方法时，这个方法将成为最终方法，无法被子类重写。但是，该方法仍然可以被继承。相当于把方法锁定，防止继承类修改它的含义。

scala复制代码class A 
{
    final void m1() 
    {
        System.out.println("This is a final method.");
    }
}

class B extends A 
{
    void m1()
    { 
        // COMPILE-ERROR! Can't override.
        System.out.println("Illegal");
    }
}

需要注意的是：当一个方法被private修饰，这会隐式的指定为final，这也会使得子类无法覆盖此方法，可以对private方法增加final修饰。

final修饰类

final修改类时，该类成为最终类，无法被继承。一旦类被final修饰，即代表final类中的所有成员变量和方法都会隐式的final。

arduino复制代码public final class FinalDemo {
    public int a;
}
public class InheritFinalDemo Extends FinalDemo{
    public int b;
}

会遇到如下提示：

Cannot inherit from final class

final相关

static 和 final

1、static强调的是该数据只存在一份，且是属于类的，不是属于对象。

2、final强调该数据不可变，且是属于对象的。

arduino复制代码 private final double A = 3.14D;
 private static double b = 3.14D;
 public static void main(String[] args) {
 new C();
 new C();
 new C();
 }
}

面代码运行后，A会存在3个，但b只有一个。

final，finally和finalize

finally是在异常处理时配合try-catch执行清理操作，需要清理的资源包括：打开的文件或网络连接等，它会把内存之外的资源恢复到他们的初始状态。无论try中是否有异常出现，finally里的操作都会被执行。
finalize这是Object基类的一个方法，垃圾收集器在将对象清除出内存之前调用的清理资源方法，且此方法只会被系统调用一次，其实finalize能做的工作，try-finally能做的更好。

总结

final关键字主要用在三个地方：变量、方法、类。

如果是基本数据类型的变量，则其数值一旦在初始化之后便不能更改;如果是引用类型的变量，则在对其初始化之后便不能再让其指向另一个对象。

当用final修饰一个类时，表明这个类不能被继承。final类中的所有成员方法都会被隐式地指定为final方法。

final修饰方法时会把方法锁定，以防任何继承类修改它的含义。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

7年Java老后端不能说的秘密：怎么样更好的优化Redis性

发表于 2021-10-31

大家好，我是一名Java后端程序员，每天开心的撸CRUD;

你猜这次我又要写个啥没有卵用的知识点呢?

不好意思，问的稍微有点早了，啥提示都没给，咋猜呢，对吧?

今天早上老板把我叫到办公室，对我说，“公司最近接了个电商小程序单子，你和王二狗，张SD参与下需求分析和设计，然后下个月开发，3个月内完成测试，上线交付”。

WC，WC，WC

。。。。。。。

“老板，老板，我没学过微信小程序，我是个Java后端程序员，你再招一个前端微信小程序开发吧”，我很低声的跟老板说。

老板很大声的吼道，“不会的东西，不会自己学吗?招新不要钱吗?你知道今年行情有多差吗，接单子容易吗?不想干就G?”

我平很想发火怼老板，但是突然想到;

上有农村年迈父母，下有襁褓小儿，媳妇还辞职在出租房带孩子。

我就低声回复：“噢，噢，好的，好的，我学。”

屌丝的人生就是这样，总得向生活低头。

努力学习吧!!!等我技术牛逼了，把老板炒了。

下面直接上超重量级干货：

回归正题：怎么样更好的优化Redis性能?

一、优化的一些建议

1、尽量使用短的key

当然在精简的同时，不要为了key的“见名知意”。对于value有些也可精简，比如性别使用0、1。

2、避免使用keys

keys , 这个命令是阻塞的，即操作执行期间，其它任何命令在你的实例中都无法执行。当redis中key数据量小时到无所谓，数据量大就很糟糕了。所以我们应该避免去使用这个命令。可以去使用SCAN,来代替。

3、在存到Redis之前先把你的数据压缩下

redis为每种数据类型都提供了两种内部编码方式，在不同的情况下redis会自动调整合适的编码方式。

4、设置key有效期

我们应该尽可能的利用key有效期。比如一些临时数据(短信校验码)，过了有效期Redis就会自动为你清除!

5、选择回收策略(maxmemory-policy)

当Redis的实例空间被填满了之后，将会尝试回收一部分key。根据你的使用方式，强烈建议使用 volatile-lru(默认) 策略——前提是你对key已经设置了超时。但如果你运行的是一些类似于 cache 的东西，并且没有对 key 设置超时机制，可以考虑使用 allkeys-lru 回收机制，具体讲解查看。maxmemory-samples 3 是说每次进行淘汰的时候会随机抽取3个key 从里面淘汰最不经常使用的(默认选项)。

arduino复制代码maxmemory-policy 六种方式 :
volatile-lru：只对设置了过期时间的key进行LRU（默认值）
allkeys-lru ： 是从所有key里 删除 不经常使用的key
volatile-random：随机删除即将过期key
allkeys-random：随机删除
volatile-ttl ： 删除即将过期的
noeviction ： 永不过期，返回错误

6、使用bit位级别操作和byte字节级别操作来减少不必要的内存使用

1 2	arduino复制代码bit位级别操作：GETRANGE, SETRANGE, GETBIT and SETBIT byte字节级别操作：GETRANGE and SETRANGE

7、尽可能地使用hashes哈希存储

8、当业务场景不需要数据持久化时，关闭所有的持久化方式可以获得最佳的性能

数据持久化时需要在持久化和延迟/性能之间做相应的权衡.

9、想要一次添加多条数据的时候可以使用管道

10、限制redis的内存大小
(64位系统不限制内存，32位系统默认最多使用3GB内存)

数据量不可预估，并且内存也有限的话，尽量限制下redis使用的内存大小，这样可以避免redis使用swap分区或者出现OOM错误。(使用swap分区，性能较低，如果限制了内存，当到达指定内存之后就不能添加数据了，否则会报OOM错误。可以设置maxmemory-policy，内存不足时删除数据)

11、SLOWLOG [get/reset/len]

1
2

lua复制代码slowlog-log-slower-than 它决定要对执行时间大于多少微秒(microsecond，1秒 = 1,000,000 微秒)的命令进行记录。
slowlog-max-len 它决定 slowlog 最多能保存多少条日志，当发现redis性能下降的时候可以查看下是哪些命令导致的。

二、管道测试

redis的管道功能在命令行中没有，但是redis是支持管道的，在java的客户端(jedis)中是可以使用的：

示例代码：

ini复制代码//注：具体耗时，和自身电脑有关(博主是在虚拟机中运行的数据)
/**
 * 不使用管道初始化1W条数据
 * 耗时：3079毫秒
 * @throws Exception
 */
@Test
public void NOTUsePipeline() throws Exception {
    Jedis jedis = JedisUtil.getJedis();
    long start_time = System.currentTimeMillis();
    for (int i = 0; i < 10000; i++) {
        jedis.set("aa_"+i, i+"");
    }
    System.out.println(System.currentTimeMillis()-start_time);
}
 
/**
 * 使用管道初始化1W条数据
 * 耗时：255毫秒
 * @throws Exception
 */
@Test
public void usePipeline() throws Exception {
    Jedis jedis = JedisUtil.getJedis();
 
    long start_time = System.currentTimeMillis();
    Pipeline pipelined = jedis.pipelined();
    for (int i = 0; i < 10000; i++) {
        pipelined.set("cc_"+i, i+"");
    }
    pipelined.sync();//执行管道中的命令
    System.out.println(System.currentTimeMillis()-start_time);
}

hash的应用

示例：我们要存储一个用户信息对象数据，包含以下信息： key为用户ID，value为用户对象(姓名，年龄，生日等)如果用普通的key/value结构来存储，主要有以下2种存储方式：

1、将用户ID作为查找key,把其他信息封装成一个对象以序列化的方式存储缺点：增加了序列化/反序列化的开销，引入复杂适应系统(Complex adaptive system)修改其中一项信息时，需要把整个对象取回，并且修改操作需要对并发进行保护。

2、用户信息对象有多少成员就存成多少个key-value对虽然省去了序列化开销和并发问题，但是用户ID为重复存储。

Redis提供的Hash很好的解决了这个问题，提供了直接存取这个Map成员的接口。Key仍然是用户ID, value是一个Map，这个Map的key是成员的属性名，value是属性值。( 内部实现：Redis Hashd的Value内部有2种不同实现，Hash的成员比较少时Redis为了节省内存会采用类似一维数组的方式来紧凑存储，而不会采用真正的HashMap结构，对应的value redisObject的encoding为zipmap,当成员数量增大时会自动转成真正的HashMap,此时encoding为ht )。

Instagram内存优化

Instagram可能大家都已熟悉，当前火热的拍照App，月活跃用户3亿。四年前Instagram所存图片3亿多时需要解决一个问题：想知道每一张照片的作者是谁(通过图片ID反查用户UID)，并且要求查询速度要相当的块，如果把它放到内存中使用String结构做key-value:

1
2
3

arduino复制代码HSET "mediabucket:1155" "1155315" "939"
HGET "mediabucket:1155" "1155315"
"939"

测试：1百万数据会用掉70MB内存，3亿张照片就会用掉21GB的内存。当时(四年前)最好是一台EC2的 high-memory 机型就能存储(17GB或者34GB的，68GB的太浪费了),想把它放到16G机型中还是不行的。

Instagram的开发者向Redis的开发者之一Pieter Noordhuis询问优化方案，得到的回复是使用Hash结构。具体的做法就是将数据分段，每一段使用一个Hash结构存储. 由于Hash结构会在单个Hash元素在不足一定数量时进行压缩存储，所以可以大量节约内存。这一点在上面的String结构里是不存在的。而这个一定数量是由配置文件中的hash-zipmap-max-entries参数来控制的。经过实验，将hash-zipmap-max-entries设置为1000时，性能比较好，超过1000后HSET命令就会导致CPU消耗变得非常大。

1
2
3

arduino复制代码HSET "mediabucket:1155" "1155315" "939"
HGET "mediabucket:1155" "1155315"
"939"

测试：1百万消耗16MB的内存。总内存使用也降到了5GB。当然我们还可以优化，去掉mediabucket:key长度减少了12个字节。

1
2
3

arduino复制代码HSET "1155" "315" "939"
HGET "1155" "315"
"939"

三、优化案例

1、修改linux中TCP监听的最大容纳数量

1	vbnet复制代码/proc/sys/net/core/somaxconn is set to the lower value of 128.

在高并发环境下你需要一个高backlog值来避免慢客户端连接问题。注意Linux内核默默地将这个值减小到/proc/sys/net/core/somaxconn的值，所以需要确认增大somaxconn和tcp_max_syn_backlog两个值来达到想要的效果。 echo 511 > /proc/sys/net/core/somaxconn 注意：这个参数并不是限制redis的最大链接数。如果想限制redis的最大连接数需要修改maxclients，默认最大连接数为10000

2、修改linux内核内存分配策略

1
2
3

vbnet复制代码错误日志：WARNING overcommit_memory is set to 0! Background save may fail under low memory condition.
To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or
run the command 'sysctl vm.overcommit_memory=1

redis在备份数据的时候，会fork出一个子进程，理论上child进程所占用的内存和parent是一样的，比如parent占用的内存为8G，这个时候也要同样分配8G的内存给child,如果内存无法负担，往往会造成redis服务器的down机或者IO负载过高，效率下降。所以内存分配策略应该设置为 1(表示内核允许分配所有的物理内存，而不管当前的内存状态如何)。内存分配策略有三种可选值：0、1、2。 0，表示内核将检查是否有足够的可用内存供应用进程使用;如果有足够的可用内存，内存申请允许;否则，内存申请失败，并把错误返回给应用进程。 1，不管需要多少内存，都允许申请。 2，只允许分配物理内存和交换内存的大小(交换内存一般是物理内存的一半)。

3、关闭Transparent Huge Pages(THP)

THP会造成内存锁影响redis性能，建议关闭

bash复制代码Transparent HugePages ：用来提高内存管理的性能
Transparent Huge Pages在32位的RHEL 6中是不支持的
执行命令 echo never > /sys/kernel/mm/transparent_hugepage/enabled
把这条命令添加到这个文件中/etc/rc.local

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Redis实现分布式锁 Redis

发表于 2021-10-31

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动。

Redis

c语言开发的基于内存的数据库，读写速度很快，因此广泛用于缓存方向。

分布式锁

我们在系统中修改已有数据时，需要先读取，然后进行修改保存，此时很容易遇到并发问题。由于修改和保存不是原子操作，在并发场景下，部分对数据的操作可能会丢失。在单服务器系统我们常用本地锁来避免并发带来的问题，然而，当服务采用集群方式部署时，本地锁无法在多个服务器之间生效，这时候保证数据的一致性就需要分布式锁来实现。

redis原生命令实现

使用redis实现分布式锁，主要是使用redis的SETNX命令(set if not exist)

加锁命令：SETNX key value，当键不存在的时候设置键并返回成功，否者返回失败，key是锁的唯一标识，一般按照业务内容来命名
解锁命令：DEL key，通过删除键值对来释放锁，以便其他线程来使用加锁命令获取锁
锁超时：EXPIRE key timeout设置key的超时时间，当锁没有被线程显式的释放时，会在达到超时时间后，自动删除锁，避免死锁。

简单的加锁代码实现如下：

arduino复制代码public boolean tryLock(String key,String requset,int timeout) {
    Long result = jedis.setnx(key, requset);
    // result = 1时，设置成功，否则设置失败
    if (result == 1L) {
      // 设置失效时间
        return jedis.expire(key, timeout) == 1L;
    } else {
        return false;
    }
}

上述代码有一些问题：

1.setnx 和 expire 两个操作是非原子性的

如果执行setnx命令设置锁成功，在执行expire命令设置失效时间时服务器宕机或者重启等其他问题导致了expire命令没有执行成功，此时锁没有设置超时时间，会有死锁的可能。

利用lua脚本将这两个操作原子化:

ini复制代码public boolean tryLock_with_lua(String key, String UniqueId, int seconds) {
    String lua_scripts = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
            "redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";
    List<String> keys = new ArrayList<>();
    List<String> values = new ArrayList<>();
    keys.add(key);
    values.add(UniqueId);
    values.add(String.valueOf(seconds));
    Object result = jedis.eval(lua_scripts, keys, values);
    //判断是否成功
    return result.equals(1L);
}

2.锁误解除

线程A获取锁，设置超时时间后，A线程的执行时间超过了超时时间，在超时时间达到后，自动释放锁，此时线程B获取到锁，当A执行完成之后，手动的将锁释放了，而此时释放的是B获取的锁，从而产生了锁误解除的问题

针对锁误解除的问题，我们可以在设置key的时候设置对应的value，value可以看成是获取锁的线程或者线程的唯一标识，可以使用uuid来作为唯一标识，在删除锁之前校验key对应的value与线程持有的value是否相同，从而避免删除了不是自己持有的锁。

3.超时解锁导致并发执行

线程A获取锁后开始执行，但是执行时间超过了锁的超时时间，此时会自动释放锁，线程B获取锁开始执行，此时就会导致线程A和线程B同时执行。

A和B两个线程并发执行解决方法：

将过期时间设置的足够长，保证代码能够在过期时间内执行完成
为拿到锁的线程设置守护线程，给要过期但是未释放的锁增加有效时间

4.不可重入

当线程在持有锁的情况下，再次请求该锁。一个锁支持在一个线程多次加锁，那么这个锁就是可重入的。反之，如果一个不可重入的锁已经被持有的线程再次加锁，那么再次加锁会失败。Redis可以对锁的重入进行计数，在加锁的时候+1，在解锁的时候-1，当技术归于0的时候，锁释放。

如下是使用本地缓存ThreadLocal的简单实现：

vbnet复制代码private static ThreadLocal<Map<String, Integer>> LOCKERS = ThreadLocal.withInitial(HashMap::new);
// 加锁
public boolean lock(String key) {
  Map<String, Integer> lockers = LOCKERS.get();
  if (lockers.containsKey(key)) {
    lockers.put(key, lockers.get(key) + 1);
    return true;
  } else {
    if (SET key uuid NX EX 30) {
      lockers.put(key, 1);
      return true;
    }
  }
  return false;
}
// 解锁
public void unlock(String key) {
  Map<String, Integer> lockers = LOCKERS.get();
  if (lockers.getOrDefault(key, 0) <= 1) {
    lockers.remove(key);
    DEL key
  } else {
    lockers.put(key, lockers.get(key) - 1);
  }
}

使用redis的 Map 数据结构在设置key的同时，计入重入次数

5.无法等待锁释放

以上的方式都是直接返回失败或者成功的结果的，如果客户端可以等待锁释放就不能使用了

可以通过客户端轮询的方式解决这个问题，当未获取到锁时，等待一段时间后重新获取结果，直到获取到锁或者等待超时。这种方法比较消耗服务器的资源，在并发量较大的时候效率较低。
使用Redis的发布订阅功能，在获取锁失败以后，订阅释放锁的信息，当锁被释放的时候，发送释放锁的信息

Redisson实现

线程获取锁的时候执行的锁lua脚本，保证了原子性
线程获取锁失败的时候，会一直通过while循环尝试获取锁，直到获取成功，再执行lua脚本(这里也包含了等待时间)
支持watch doc自动延期，这一点是针对上面锁说的超时解锁导致并发执行的情况，这里watch dog在后台开启了一个线程，不断延长key的生存时间。但是这个相当于监控线程的watch dog会对性能有一定的影响
实现了可重入锁的机制

1. redis本身的存储数据结构支持Map
2. Map的key值可以表示当前的线程信息，value可以用来记录重入的次数

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

【MP】还在用 QueryWrapper 吗？

发表于 2021-10-31

前言

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动。我们都知道在 MP 中 QueryWrapper（LambdaQueryWrapper）和 UpdateWrapper（LambdaUpdateWrapper）的父类 AbstractWrapper 用于生成 sql 的 where 条件，但是 Wrapper 这么多你只会用 QueryWrapper 可远远不够的啊！各种高级骚操作必须学起来，结合案例代码轻松驾驭各种用法，顺便梳理一些常用的条件构造器及使用的注意事项，闲话少叙，直接进入正题。

前期准备

准备两张表以及对应的实体类对象

tb_province 省份表

sql复制代码CREATE TABLE `tb_province` (
  `pid` int(11) NOT NULL AUTO_INCREMENT COMMENT '省份编号',
  `province` char(4) DEFAULT NULL COMMENT '省份名',
  `abbr` varchar(3) DEFAULT NULL COMMENT '省份的简称',
  `area` int(11) DEFAULT NULL COMMENT '省份的面积（km²）',
  `population` decimal(10,2) DEFAULT NULL COMMENT '省份的人口（万）',
  `attraction` varchar(50) DEFAULT NULL COMMENT '省份的著名景点',
  `postcode` varchar(10) DEFAULT NULL COMMENT '省份的省会邮政编码',
  PRIMARY KEY (`pid`),
  KEY `postcode` (`postcode`),
  CONSTRAINT `tb_province_ibfk_1` FOREIGN KEY (`postcode`) REFERENCES `tb_capital` (`postcode`)
) ENGINE=InnoDB AUTO_INCREMENT=35 DEFAULT CHARSET=utf8;

省份表对应的实体类

java复制代码@TableName(value = "tb_province")
@Data
public class Province implements Serializable {

    private static final long serialVersionUID = 1L;
    
    /**
    * 省份编号
    */
    @TableId(value = "pid", type = IdType.AUTO)
    private Integer pid;

    /**
    * 省份名
    */
    private String province;

    /**
    * 省份的简称
    */
    private String abbr;

    /**
    * 省份的面积
    */
    private Integer area;

    /**
    * 省份的人口
    */
    private BigDecimal population;

    /**
    * 省份的著名景点
    */
    private String attraction;

    /**
    * 通过省份的省会邮政编码关联省会信息
    */
    @TableField(exist = false)
    private Capital capital;
}

tb_capital 省会表，通过省份表的邮政编码关联

sql复制代码CREATE TABLE `tb_capital` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键id',
  `postcode` varchar(10) NOT NULL COMMENT '省会的邮政编码',
  `city` varchar(4) DEFAULT NULL COMMENT '省会名',
  `nickname` varchar(10) DEFAULT NULL COMMENT '省会的别名',
  `climate` varchar(20) DEFAULT NULL COMMENT '省会的气候条件',
  `carcode` varchar(5) DEFAULT NULL COMMENT '省会的车牌号',
  PRIMARY KEY (`id`,`postcode`) USING BTREE,
  KEY `postcode` (`postcode`),
  CONSTRAINT `tb_capital_ibfk_1` FOREIGN KEY (`postcode`) REFERENCES `tb_province` (`postcode`)
) ENGINE=InnoDB AUTO_INCREMENT=36 DEFAULT CHARSET=utf8;

省会表对应的实体类

java复制代码@TableName(value = "tb_capital")
@Data
public class Capital implements Serializable {
    
    private static final long serialVersionUID = 1L;
    
    /**
    * 主键id
    */
    private Integer id;

    /**
    * 省会的邮政编码
    */
    private String postcode;

    /**
    * 省会名
    */
    private String city;

    /**
    * 省会的别名
    */
    private String nickname;

    /**
    * 省会的气候条件
    */
    private String climate;

    /**
    * 省会的车牌号
    */
    private String carcode;
}

代码演练

OK，一切准备就绪后开始撸代码。

普通 QueryWrapper

先来个简单的，查询省份表中的某一条记录，你可能会用 QueryWrapper 这么写：

java复制代码// 查询江西省基本信息
QueryWrapper<Province> wrapper = new QueryWrapper<>();
wrapper.eq("province", "江西省");
Province province = provinceMapper.selectOne(wrapper);
System.out.println(province);

执行后，控制台中可以看见如下语句：

sql复制代码==>  Preparing: SELECT pid,province,abbr,area,population,attraction FROM tb_province WHERE (province = ?) 
==> Parameters: 江西省(String)
<==    Columns: pid, province, abbr, area, population, attraction
<==        Row: 10, 江西省, 赣, 166900, 4666.10, 庐山、鄱阳湖、滕王阁
<==      Total: 1

QueryWrapper 查询条件包装类使用 eq（equal）方法将传入的第一个参数（数据库表中的列名）和第二个参数（条件值）划上等号，然后调用 mapper 接口继承 BaseMapper 下来的 selectOne() 方法，传入 Wrapper 将查询条件加上，于是就得到了上面的 SQL 语句。

最终打印结果为：

1	console复制代码Province [Hash = 629321967, pid=10, province=江西省, abbr=赣, area=166900, population=4666.10, attraction=庐山、鄱阳湖、滕王阁, capital=null]

capital=null ！！因为数据库表字段不对应，使用注解排除了非表字段。

那如何在查询省份信息的时候，将其所属的省会信息塞进实体对象中，从而得到一个比较详细的省份详细信息呢？

两种做法：

Province 类中删除被注解标注的 capital 属性，加上外键通过这个字段关联到 Capital 实体，做一个二次查询然后再赋值。
一个省份对应一个省会，是一对一关系，所以我们可以在对应 xml 文件中写 resultMap 结果集映射，将外键值通过映射查询返回的结果映射到 capital 属性上。

这里使用第二种做法：

ProvinceMapper.xml：

xml复制代码<resultMap id="BaseResultMap" type="com.xx.xxx.entity.Province">
  <id column="pid" jdbcType="INTEGER" property="pid" />
  <result column="province" jdbcType="CHAR" property="province" />
  <result column="abbr" jdbcType="VARCHAR" property="abbr" />
  <result column="area" jdbcType="INTEGER" property="area" />
  <result column="population" jdbcType="DECIMAL" property="population" />
  <result column="attraction" jdbcType="VARCHAR" property="attraction" />
  <association property="capital" javaType="com.xx.xxx.entity.Capital" column="postcode"
               select="com.xx.xxx.mapper.CapitalMapper.selectAllByPostcode">
  </association>
</resultMap>

CapitalMapper.xml 外键值映射查询 SQL ：

xml复制代码<select id="selectAllByPostcode" parameterType="map" resultMap="BaseResultMap">
  select 
  <include refid="Base_Column_List" />
  from tb_capital
  where postcode = #{postcode,jdbcType=VARCHAR}
</select>

只要 resultMap="BaseResultMap" ，那么你查询的省份信息就包含省会相关信息了。

java复制代码Province province = provinceMapper.selectByProvinceName("浙江省");
System.out.println(province); // Province [Hash = 120157876, pid=7, province=浙江省, abbr=浙, area=105500, population=5850.00, attraction=西湖、乌镇、千岛湖, capital=Capital [Hash = 1919497442, id=14, postcode=310000, city=杭州, nickname=临安, climate=亚热带季风气候, carcode=浙A]]

/* 执行后的 SQL 语句
==>  Preparing: select pid, province, abbr, area, population, attraction, postcode from tb_province where province=? 
==> Parameters: 浙江省(String)
<==    Columns: pid, province, abbr, area, population, attraction, postcode
<==        Row: 7, 浙江省, 浙, 105500, 5850.00, 西湖、乌镇、千岛湖, 310000
====>  Preparing: select id, postcode, city, nickname, climate, carcode from tb_capital where postcode = ? 
====> Parameters: 310000(String)
<====    Columns: id, postcode, city, nickname, climate, carcode
<====        Row: 14, 310000, 杭州, 临安, 亚热带季风气候, 浙A
<====      Total: 1
<==      Total: 1
*/

Wrapper 支持链式编程，如果查询条件：省份名为浙江省、简称为浙且邮政编码为 310000，你可能会这么写：

java复制代码QueryWrapper<Province> eq = new QueryWrapper<Province>()
        .eq("province", "浙江省")
        .eq("abbr", "浙")
        .eq("postcode", "310000");
Province province = provinceMapper.selectOne(eq);
System.out.println(province);

其实可以不用接连写三个 eq() ，直接写一个 allEq() 即可：

1 2	java复制代码QueryWrapper<Province> eq = new QueryWrapper<Province>() .allEq({"province": "浙江省", "abbr": "浙", "postcode": "310000"}, true);

第一个参数接收一个 Map ，key 为数据库字段名， value为字段值；

第二个参数接收一个布尔值，可以不传，默认为 true，为 true 则在 map 的 value 为 null 时调用 isNull 方法，为 false 时则不将 value 为 null 的字段作为查询条件。

链式 Lambda 操作

通过上面代码实例可以明显发觉，每次都要自己写 column_name ，一旦写错立马报错，而且写固定列名损害了代码的健壮性，比较死板。

所以使用函数式接口，实现链式查询就非常有必要了！

查询省份名带有 ”江“ 、人口超过 2000 万或省份面积在 10w~25w 平方千米的省份信息，按照人口数量降序显示。

java复制代码// 在 QueryWrapper 中是获取的是 LambdaQueryWrapper
LambdaQueryWrapper<Province> eq = new LambdaQueryWrapper<Province>()
        .like(Province::getProvince, "江")
        .gt(Province::getPopulation, 2000)
        .or()
        .between(Province::getArea, 100000, 250000)
        .orderByDesc(Province::getPopulation);
List<Province> provinces = provinceMapper.selectList(eq);
Optional.ofNullable(provinces).ifPresent(p -> p.forEach(System.out::println));

注意：不调用 or() 则默认为使用 and 连接。

执行后的 SQL 语句：

sql复制代码==>  Preparing: SELECT pid,province,abbr,area,population,attraction FROM tb_province WHERE (province LIKE ? AND population > ? OR area BETWEEN ? AND ?) ORDER BY population DESC 
==> Parameters: %江%(String), 2000(Integer), 100000(Integer), 250000(Integer)
<==    Columns: pid, province, abbr, area, population, attraction
<==        Row: 12, 广东省, 粤, 179725, 11521.00, 丹霞山、华侨城、白云山
<==        Row: 19, 山东省, 鲁, 157900, 10070.21, 泰山、沂蒙山、蓬莱阁
<==        Row: 11, 河南省, 豫, 167000, 9640.00, 少林寺、龙门石窟、殷墟
<==        Row: 6, 江苏省, 苏, 107200, 8070.00, 中山陵、花果山、三台山
<==        Row: 1, 河北省, 冀, 188800, 7591.97, 白洋淀、避暑山庄、北戴河
<==        Row: 17, 湖南省, 湘, 211800, 6918.38, 张家界、岳阳楼、衡山
<==        Row: 8, 安徽省, 皖, 140100, 6365.90, 黄山、九华山、天堂寨
<==        Row: 18, 湖北省, 鄂, 185900, 5927.00, 黄鹤楼、神农架、长江三峡
<==        Row: 7, 浙江省, 浙, 105500, 5850.00, 西湖、乌镇、千岛湖
<==        Row: 13, 广西, 桂, 237600, 4960.00, 桂林山水、银滩、青秀山
<==        Row: 10, 江西省, 赣, 166900, 4666.10, 庐山、鄱阳湖、滕王阁
<==        Row: 3, 辽宁省, 辽, 148600, 4351.70, 沈阳东陵、大连星海广场
<==        Row: 9, 福建省, 闽, 124000, 3973.00, 鼓浪屿、武夷山、涠洲岛
<==        Row: 20, 陕西省, 陕, 205600, 3876.21, 兵马俑、华山、黄帝陵
<==        Row: 5, 黑龙江省, 黑, 473000, 3751.30, 北极村、扎龙湿地、五大连池
<==        Row: 2, 山西省, 晋, 156700, 3729.22, 五台山、平遥古城、云冈石窟
<==        Row: 14, 贵州省, 黔或贵, 176167, 3622.95, 黄果树瀑布、梵净山、万峰林
<==        Row: 4, 吉林省, 吉, 187400, 2690.73, 长白山、净月潭、高句丽王陵
<==      Total: 18

按照气候进行分组，并筛选出别名在三个汉字及以上的直辖市。

第一个入参 boolean condition 表示该条件是否加入生成的 SQL 中，默认为 true 。例如：query.like(StringUtils.isNotBlank(name), Entity::getName, name) .eq(age!=null && age >= 0, Entity::getAge, age)

java复制代码List<Capital> capitals = new LambdaQueryChainWrapper<Capital>(capitalMapper)
        .isNull(Capital::getCity)
        .groupBy(Capital::getClimate)
        .having(true, "length(nickname) >= 9") // 等价于 .having(true, "length(nickname) < {0}", 9) {0} 占位
        .list();

Optional.ofNullable(capitals).ifPresent(list -> {
    list.forEach(System.out::println);
});

LambdaQueryChainWrapper 链式查询 Lambda 式，使用 CapitalMapper 接口（继承 BaseMapper）初始化，链式的末尾使用 list() 方法调用 this.getBaseMapper().selectList(this.getWrapper()); 返回一个 List 结果集。

同时，最后也可通过

one() 返回 T 实体类对象
oneOpt() 返回 Optional<T> Optional 包装实体类
count() 返回 Integer 统计结果集总条数
page() 返回 Page 分页对象。

除了使用 new LambdaQueryWrapper<T>() 的方式得到一个 LambdaQueryWrapper，还可以使用 Wrappers 类调用 query() 得到一个 QueryWrapper 对象，调用 lambdaQuery() 静态方法得到一个 LambdaQueryWrapper等等

查询简称有两个及以上和不包括含有“海”的著名景点的省份下的省会信息，如果 randomBool 随机布尔值为 true 那么就再去筛选省会别名不为“林城”，否则判断一条查询 SQL 语句是否存在结果，最终的查询的记录按照主键 id 升序且只选取前两个记录。

这个案例需要使用到了两张表，可用子查询 inSql() ，第一个参数为表列名，另一个参数为 sqlString 即需传入 SQL 语句，拼接后的效果为：列 in (查询结果集) 。

java复制代码// 生成一个随机布尔值
boolean randomBool = Math.random() > 0.5d;

LambdaQueryWrapper<Capital> lambdaQueryWrapper = Wrappers.<Capital>lambdaQuery().
        inSql(Capital::getPostcode, "select postcode \n" +
                "from tb_province \n" +
                "where length(abbr) != 3 and attraction not like '%海%'")
        .func(true, wrapper -> { // condition：true，可忽略
            if (randomBool) { 
                wrapper.ne(Capital::getNickname, "林城"); 
            } else { 
                wrapper.exists("select * from tb_capital where climate = '亚热带季风性气候'"); 
            } 
        })
        .orderByAsc(Capital::getId)
        .last("limit 2");

List<Capital> capitals = capitalMapper.selectList(lambdaQueryWrapper);

Optional.ofNullable(capitals).ifPresent(list -> {
    list.forEach(System.out::println);
});

func() 方法主要作用就是要方便在出现 if...else 下调用不同方法能不断链，两种情况的不同会得到不同的 SQL 语句。

运行后，控制台打印的 SQL 语句：

sql复制代码randomBool: true
==>  Preparing: SELECT id,postcode,city,nickname,climate,carcode FROM tb_capital WHERE (postcode IN (select postcode from tb_province where length(abbr) != 3 and attraction not like '%海%') AND nickname <> ?) ORDER BY id ASC limit 2 
==> Parameters: 林城(String)
<==    Columns: id, postcode, city, nickname, climate, carcode
<==        Row: 25, 610000, 成都, 天府之国, 亚热带季风性湿润气候, 川A
<==        Row: 28, 730000, 兰州, 金城, 温带大陆性气候, 甘A
<==      Total: 2

randomBool: false
==>  Preparing: SELECT id,postcode,city,nickname,climate,carcode FROM tb_capital WHERE (postcode IN (select postcode from tb_province where length(abbr) != 3 and attraction not like '%海%') AND EXISTS (select * from tb_capital where climate = '亚热带季风性气候')) ORDER BY id ASC limit 2
==> Parameters: 
<==      Total: 0

last() 无视优化规则直接拼接到 SQL 的最后，但是要注意只能调用一次，如果多次调用以最后一次为准，且有 sql 注入的风险，需谨慎使用！！

CustomSqlSegment 入参

经常会遇到这样一个需求：写一个搜索接口，传入的参数是一个 VO 对象，里面包含的都是一些搜索字段，返回的搜索结果放到 List<VO> / page<VO> 中，这时如果使用 mapper.selectPage() 返回并不是一个包装 VO 的 Page 对象，所以这时你就需要自定义 mapper 层方法，编写 xml 文件了。

ProvinceVO 对象：

java复制代码@Data
public class ProvinceVO {

    private Integer pid;

    private String province;a

    private String abbr;
    
    ......
}

Mapper 层接口方法：

1
2
3

java复制代码List<ProvinceVO> queryPageList(Page page, @Param("provinceVO") ProvinceVO provinceVO);

Page<ProvinceVO> pageList(Page<ProvinceVO> page, @Param("provinceVO") ProvinceVO provinceVO);

mapper 接口 queryFruitList() 方法返回的是分页后的 List 集合，但是也可以是 Page 的包装对象，拿到里面的列表数据只需要调用 getRecords() 静态方法。

java复制代码Page<ProvinceVO> page = new Page<ProvinceVO>(1, 10);
ProvinceVO provinceVO = new ProvinceVO();
provinceVO.setProvince("江");
Page<ProvinceVO> provinceVOPage = provinceMapper.pageList(page, provinceVO);

System.out.println(provinceVOPage.getTotal()); // 查询到的总记录条数
System.out.println(provinceVOPage.getCurrent()); // 当前页
System.out.println(provinceVOPage.getRecords()); // 分页列表数据

// 与 provinceVOPage.getRecords() 等价
List<ProvinceVO> provinceVOPageList = fruitMapper.queryPageList(page, provinceVO);
Optional.ofNullable(provinceVOPageList).ifPresent(list -> {
    list.forEach(System.out::println);
});

如果你想使用 Wrapper 自定义 SQL 生成 where 条件，你可以使用注解的方式来书写：

1 2	java复制代码@Select("select * from tb_province ${ew.customSqlSegment}") List<ProvinceVO> getVOListByCustomSqlSegment(@Param("ew") Wrapper ew);

测试代码：

java复制代码List<ProvinceVO> provinceVOList = provinceMapper.getVOListByCustomSqlSegment(new QueryWrapper<ProvinceVO>()
        .like("province", "江"));

Optional.ofNullable(provinceVOList).ifPresent(list -> {
    list.forEach(System.out::println);
});

注意：不支持 Wrapper 内的 entity 生成 where 语句！！也就是说不能使用函数式接口代替列名。

java复制代码List<ProvinceVO> provinceVOList = provinceMapper.getVOListByCustomSqlSegment(new LambdaQueryWrapper<ProvinceVO>()
        .like(ProvinceVO::getProvince, "江"));*/

// 或者
List<ProvinceVO> provinceVOList = provinceMapper.getVOListByCustomSqlSegment(Wrappers.<ProvinceVO>lambdaQuery()
        .like(ProvinceVO::getProvince, "江"));

否则，控制台会报如下错误信息：

1
2
3

console复制代码org.apache.ibatis.builder.BuilderException:
    Error evaluating expression 'ew.customSqlSegment'.
    ... MybatisPlusException: can not find lambda cache for this entity.

除了，注解的方式，还可以在 XML 文件中书写：

Constants.WRAPPER 使用了 MP 中字符串常量池，其值为 ew ，等价于 @Param("ew")

1	java复制代码Page<ProvinceVO> getPageListByCustomSqlSegment(Page<ProvinceVO> page, @Param(Constants.WRAPPER) Wrapper wrapper);

XML 中书写 SQL 语句：

sql复制代码<select id="getPageListByCustomSqlSegment" resultType="com.xx.xxx.vo.ProvinceVO">
  select *
  from tb_province
  ${ew.customSqlSegment}
</select>

测试代码：

java复制代码Page<ProvinceVO> page = new Page<>(1, 10);

Page<ProvinceVO> provinceVOPage = provinceMapper.getPageListByCustomSqlSegment(page, new QueryWrapper<Province>()
        .like("province", "江"));

Optional.ofNullable(provinceVOPage).ifPresent(p -> {
    List<ProvinceVO> provinceVOList = p.getRecords();
    long total = p.getTotal();
    long currentPage = p.getCurrent();
    long pageSize = p.getSize();

    provinceVOList.forEach(System.out::println);

    System.out.println("总条数：" + total);
    System.out.println("当前页：" + currentPage);
    System.out.println("每页条数" + pageSize);
});

执行后，控制台可见：

console复制代码JsqlParserCountOptimize sql=select *
    from tb_province
    WHERE (province LIKE ?)
==>  Preparing: SELECT COUNT(1) FROM tb_province WHERE (province LIKE ?) 
==> Parameters: %江%(String)
<==    Columns: COUNT(1)
<==        Row: 4
==>  Preparing: select * from tb_province WHERE (province LIKE ?) LIMIT ?,? 
==> Parameters: %江%(String), 0(Long), 10(Long)
<==    Columns: pid, province, abbr, area, population, attraction, postcode
<==        Row: 5, 黑龙江省, 黑, 473000, 3751.30, 北极村、扎龙湿地、五大连池, 150000
<==        Row: 6, 江苏省, 苏, 107200, 8070.00, 中山陵、花果山、三台山, 210000
<==        Row: 7, 浙江省, 浙, 105500, 5850.00, 西湖、乌镇、千岛湖, 310000
<==        Row: 10, 江西省, 赣, 166900, 4666.10, 庐山、鄱阳湖、滕王阁, 330000
<==      Total: 4

ProvinceVO{pid=5, province='黑龙江省', abbr='黑'}
ProvinceVO{pid=6, province='江苏省', abbr='苏'}
ProvinceVO{pid=7, province='浙江省', abbr='浙'}
ProvinceVO{pid=10, province='江西省', abbr='赣'}
总条数：4
当前页：1
每页条数10

注意：${ew.customSqlSegment} 前面千万不要加 where 关键字，QueryWrapper 会附带。

结尾

撰文不易，欢迎大家点赞、评论，你的关注、点赞是我坚持的不懈动力，感谢大家能够看到这里！Peace & Love。

参考

条件构造器 | MyBatis-Plus (baomidou.com)

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

MySQL 执行计划（explain）使用详解执行计划是什

发表于 2021-10-31

本文已参与「掘力星计划」，赢取创作大礼包，挑战创作激励金。

执行计划是什么？

使用 EXPLAIN 关键字可以模拟优化器执行SQL查询语句，从而知道MySQL是
如何处理你的SQL语句的，分析你的查询语句或是表结构的性能瓶颈。

官网介绍： dev.mysql.com/doc/refman/…

前提介绍：文中所有案例 mysql 版本为 5.7.23

执行计划帮助我们完成什么事情？

表的读取顺序
数据读取操作的操作类型
哪些索引可以使用
哪些索引被实际使用
表之间的引用
每张表有多少行被优化器查询

怎么使用执行计划？

expain + SQL 语句
执行计划包含信息

执行计划包含信息解释

id

select 查询的序列号，包含一组数字，表示查询中执行 select 子句或操作表的顺序

sql复制代码use oemp;

#测试表1
CREATE TABLE `t1` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `other_column` varchar(30) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

#测试表2
CREATE TABLE `t2` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `other_column` varchar(30) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

#测试表3
CREATE TABLE `t3` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `other_column` varchar(30) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

#id 相同
explain select t2.* from t1,t2,t3 where t1.id = t2.id and t1.id = t3.id
and t3.other_column = '';

#id 不同
explain select t2.* from t2 where id = (select id from t1 where id = 
(select t3.id from t3 where t3.other_column = ''));

#id 相同和不同同时存在
explain select t2.* from (select t3.id from t3 where t3.other_column = '') s1,t2 
where s1.id = t2.id;

包含三种情况： id 相同，id 不同，id 相同和 id 不同同时存在。

id 相同

id 相同，执行结果从上而下

运行结果

id 不同

id不同如果是自查询，id 的序号会递增，id 值越大，优先级越高，越先被执行

运行结果

id 相同和 id 不同时存在

id 如果相同，可以认为是一组的，从上往下执行；
在所有组中，id 值越大，优先级越高，越先被执行；
衍生 = DERIVED

执行结果

derived_merge 是 Mysql5.7 引入的，会试图将 Derived Table (派生表，from 后的自查询) 视图引用，公用表达式（Common table expressions）与外层查询进行合并。
MySQL 5.7 不在兼容的实现方式，可以通过调整 optimizer_switch 来加以规避

1	sql复制代码set optimizer_switch='derived_merge=off';

说白了，如果设置为 on 那么就不会出现 derived_merge 行
结果如下：

select_type

包括范围： simple. primary，subquery, derived, union, union result
查询类型主要是用于区别普通查询，联合查询，子查询等复杂的查询

simple，简单的select 语句，查询中不包含自查询或者 union
primary, 查询若包含任何复杂的子部分，最外层查询则被标记为primary
subquery，在 select 或 where 列表中包含子查询
derived，在 from 列表中包含自查询被标记为 derived （衍生）MySQL 会递归执行这些自查询，把结果放在临时表中。
union，若第二个 select 出现在 union 之后，则被标记为 union. 若 union 包含在 from 子句子查询中，外层 select 将别标记为 derived
union result, 从 union 表中获取结果的 select

table

这行数据是关于那种表的

type

类型： all , index , range, ref, eq_ref, const, system ,null
type 显示的是防卫类型，是较为重要的一个指标，结果从好到坏依次是：
system > count > eq_ref > range > index > all

sytem > const > eq_ref > ref > fulltext > ref_or_null > index_merge >> unique_subquery > index_subquery > range > index > ALL

system

表只有一行记录（等于系统表），这是 const 类型的特列，平时不会出现，这个也可以忽略不计

const

1	sql复制代码explain select * from (select * from t1 where id =1) d1;

表示通过索引一次就找到了， const 用于比较 primary key 或者 unique 索引。因为只匹配一行数据，所以很快如将主键置于where 列表中， MySQL 就能将该查询转换为一个常量。

eq_ref

1	sql复制代码explain select * from t1, t2 where t1.id = t2.id;

唯一性索引扫描，对于每个索引键，表中只有一条记录与之匹配。常见于主键或唯一索引扫描.
查询案例：

image.png

ref

sql复制代码# tb_emp ddl
CREATE TABLE `tb_emp` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(30) DEFAULT NULL,
  `dept_id` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`),
) ;

#员工表添加年龄列
alter table tb_emp add column `age` int(11) default null after `name`;

#添加复合索引
create index idx_emp_name_age on tb_emp(`name`, `age`);

explain select * from tb_emp where `name` = 'z3';

非唯一性索引扫描，返回匹配某个单独值的所有行，本质上也是一种索引访问，它返回所有匹配某个单独的行，然而，它可能会找到多个符合个条件的行，所以它应该属于查找和扫描的混合体

image.png

range

1
2
3

sql复制代码explain select * from t1 where id between 1 and 3;

explain select * from t1 where id in (1, 2, 3);

只检索给定范围内的行，使用一个索引来选择行。key 列显示使用了哪个索引
一般就是你在 where 语句中出现了 between、<、>、in 等的查询
这种范围扫描索引比全表扫描要好，因为它只需要开始于索引的某个点，而结束于另一个点，不用全表扫描
案例结果：

index

1	sql复制代码explain select id from t1;

Full Index Scan ， index 于 ALL的却别，index 类型只遍历索引树，这通常比 ALL 快，因为索引文件通常比数据文件小。（也就是说虽然 all 和 index 都是读全表，但是index 是从索引中读取的，而 all 是从硬盘中读取的）
查询结果：

all

1	sql复制代码explain select * from t1;

Full Table Scan 将遍历全表找到匹配的行

备注：一般来说，得以保证查询至少达到 rang 级别，最好能达到 ref。

possible_keys

显示可能应用在这张表中的索引，一个或多个。
查询涉及到的字段上若存在索引，则该索引将被列出，但不一定被查询实际使用.

key

实际使用的索引，如果为NULL，则没有使用索引
查询中若使用了覆盖索引，则该索引仅出现在KEY列表中

sql复制代码explain select col1, col2  from t1;

create index idx_col1_col2 on t1(col1, col2);

explain select col1, col2  from t1;

案例一（加索引之前）

案例二（加索引之后）

key_len

1
2
3

sql复制代码desc t1; 
explain select * from t1 where col1 = 'ab';
explain select * from t1 where col1 = 'ab' and col2 = 'bc';

表示索引中使用的字节数，可通过该列计算查询中的使用的索引的长度，在不损失精确性的情况下，长度越短越好
key_len 显示的只为索引字段的最大可能长度，** 并非实际使用长度**。即 key_len e是更具表定义计算而得，不是通过表内检索出的。
查询结果：

总结：条件越多，付出的代价越大，key_len 的长度也就越大，建议在一定条件的情况下，key_len 越短，效率越高。

Rows

根据表统计信息及索引选用情况，大致估算出找到所需的记录所需读取的行数

filtered

Extra

包含不适合其他列中显示但十分重要的额外信息
id, select_type, table, type , possible_keys, key, key_len, ref, rows, Extra

1. Using filesort

文件排序

2. Using temporary

sql复制代码explain select col2 from t1 where col1 in ('ab', 'ac', 'as') group by col2 \G;

explain select col2 from t1 where col1 in ('ab', 'ac', 'as') 
group by col1, col2, col3 \G;

使用了临时表保存中间结果， MySQL 在对查询结果排序时使用临时表。
常见于排序 order by 和分组查询 group by 。
例子：

3. Using index

1
2
3

sql复制代码explain select col2 from  t1 where col1=100;

explain select col1, col2 from  t1;

表示相应的 select 操作使用了覆盖索引 (Covering Index), 避免了访问表的数据行，效率不错～
如果同时出现 using where , 表示索引被用来执行索引键值的查找；
如果没有同时出现 using where , 表明索引引用来读取数据而非执行查找动作。
例子：

覆盖索引（Covering Index）

覆盖索引（Covering Index）, 一说为索引覆盖
理解方式一：就是 select 的数据列只用从索引中就能取得，不必读取数据行， MySQL 可以利用你索引返回 select 列表的字段，而不必根据索引再次读取数据文件，换句话说查询列要被所建的索引覆盖
理解方式二：索引是高效找到的行的一个方法，但是一般数据库也能使用索引找到一个列的数据，因此它不必读取整个行，毕竟索引叶子节点存储了他们索引的数据；当能通过读取索引就可以得到想要的数据，那就不需要读取行了。一个索引包含了（或覆盖了）满足查询结果的数据就叫做覆盖索引。
注意： 1. 如果要使用覆盖索引，一定要注意 select 列表汇总只取出需要的列，不可 select * ； 2. 因为如果将所有字段一起做索引将会导致索引文件过大，查询性能下降。

4. Using Where

表明使用了 where 过滤

5. using join buffer

使用了链接缓存

6. impossible where

1	sql复制代码explain select * from t1 where 1=2;

where 子句的值总是 false , 不能用来获取任何元组

7. select tbale optimized away

在没有 GROUPBY 子句的情况下，基于索引优化 MIN/MAX 操作或者对于 MyISAM 存储引擎优化 COUT（*）操作不必等到执行阶段再进行计算，查询执行计划生成的阶段即完成优化。

8. distinct

优化 distinct 操作在找到第一匹配的元祖后立即停止找相同值的动作。

举个例子

例子描述：

sql复制代码explain select d1.name, (select id from t3) d2 from 
(select id, name from t1 where other_column = '') d1 
union 
(select name, id from t2);

查询结果：

案例解析：

第一行（执行顺序4）：id 列为1 ，表示 union 的第一个 select , select_type 的 primary 表表示该查询为外层查询， table
列被标记为 , 表示查询结果来自一个衍生表，其中 derived3 中的 3 代表查询衍生自第三个 select 查询，即 id 为 3 的 select [select d1.name … ]
第二行（执行顺序为2）：id 为 3 ，是整个查询中第三个 select 的一部分，因查询包含在from 中，所以为derived 。【select id, name from where other_column = ‘’】
第三行（执行顺序为3）：select 列表中的子查询 select_type 为 subquery , 为整个查询中的第二个 select . [select id from t3]
第四行（执行顺序为1）：select_type 为 union ，说明第四个 select 是 unin 里的第二个 select , 最先执行【select name ,id from t2】
第五行（执行顺序为5）：代表 union 的临时表中读取行的阶段， table 列的 <union , 1, 4> 表示用第一个和第四个 select 结果进行union 操作。【两个结果 union 操作】

参考资料

「欢迎在评论区讨论，掘金官方将在掘力星计划活动结束后，在评论区抽送100份掘金周边，抽奖详情见活动文章」

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

疫情让我们学会了什么

发表于 2021-10-31

前言

事情开始于2019年12月8日，官方通报首例不明原因肺炎患者病历。

到现在时间已经过去了将近2年，我们经历了封城、隔离、居家办公，我们有过恐慌，有过感动，

我们曾被无私的医生、护士、军人、志愿者等等而感动，他们的“病情不退誓不回”一次次让我们落泪。

我们经历了火神医院十天完工的奇迹，我们经历的太多太多了。

每当有难的时候，你就能体会的你所在的国家是如此的强大，面对疫情我们毫不逊色于某些发达国家。

此生无悔入华夏，来世还做华夏人！

一、学习

现在是2021年10月31日，疫情已经取得了阶段性胜利。

身为程序员的我，能做到的是不为国家添麻烦。

所以我平时也不出去浪，在家学习学习，提升一下自己也挺不错。

之前看ReentrantLock源码的时候涉及到AQS,里边用到了模板方法模式，所以我打算今天输出一篇模板方法模式的文章。

疫情期间有段时间是全部居家隔离，于是有人在学习做饭，有人在减肥，有人在造娃。

我们以此为例讲一下模板方法。

file

代码实现：

AbstractRegulations

java复制代码package test;


/**
 * @author 木子的昼夜编程
 */
public abstract class AbstractRegulations {

    // 每个人都有自己的名字
    String name;
    public AbstractRegulations(){}
    public AbstractRegulations(String name) {
        this.name = name;
    }
    // 国家指定了你的行动范围  和所有大概活动
    // 但是不管你在家干什么
    public void doWhat(){
        System.out.println(name+"早上9点小区带好口罩，楼下领菜。");
        stayAtHome();
        System.out.println(name+"晚上8点自家门口带好口罩，等待志愿者测体温。");
    }

    protected void stayAtHome() {
        throw new UnsupportedOperationException();
    }
}

XiaoMingRegulations

java复制代码package test;

/**
 * @author 木子的昼夜编程
 */
public class XiaoMingRegulations extends AbstractRegulations{
    public XiaoMingRegulations(){}
    public XiaoMingRegulations(String name){
        super(name);
    }
    // 小明自定义自己要干的事情
    @Override
    protected void stayAtHome() {
        System.out.println("我是小明，我在家吃饭，吃的胖胖的。");
    }
}

XiaoQiangRegulations

java复制代码package test;

/**
 * @author 木子的昼夜编程
 */
public class XiaoQiangRegulations extends AbstractRegulations{
    public XiaoQiangRegulations(){}
    public XiaoQiangRegulations(String name){
        super(name);
    }
    // 小强自定义自己干的事情
    @Override
    protected void stayAtHome() {
        System.out.println("我是小强，我在家练腰，练得杠杠的。");
    }
}

Test

java复制代码package test;

/**
 * @author 木子的昼夜编程
 */
public class Test {
    public static void main(String[] args) {
        new XiaoMingRegulations("小明").doWhat();
        System.out.println("---------------华丽丽的分割线-------------------");
        new XiaoQiangRegulations("小强").doWhat();
    }
}

输出：

tex复制代码小明早上9点小区带好口罩，楼下领菜。
我是小明，我在家吃饭，吃的胖胖的。
小明晚上8点自家门口带好口罩，等待志愿者测体温。
---------------华丽丽的分割线-------------------
小强早上9点小区带好口罩，楼下领菜。
我是小强，我在家练腰，练得杠杠的。
小强晚上8点自家门口带好口罩，等待志愿者测体温。

二、唠唠

看过我之前文章的人都知道，AQS也是用了模板方法模式。

我们稍微回顾一下AQS的代码

java复制代码public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    // 尝试获取凭证
    public final void acquire(int arg) {
        // 调用tryAcquire 尝试获取一次
        if (!tryAcquire(arg) &&
            // 获取失败就放到队列中
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    // 可以看到tryAcquire是没有实现的，需要子类来自己实现
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
}

所以当我们看到这里的时候不要懵，这种的一般都是子类实现，直接看她对应的实现类就可以了。

file

三、更多

还要更多？今天么有了，专门找了一个简单的知识点进行文章的输出。

还有点儿工作没有完成，要在家加加班，做一下。

打工人，打工魂，打工的都是人上人，今天搬砖你不狠，明天地位就不够稳。

file

再见各位打工人。

欢迎关注公众号：木子的昼夜编程

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

Terraform 基础设施即代码测试流程（二）

发表于 2021-10-31

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动。

紧跟上一篇文章《# Terraform | 基础设施即代码测试流程（一）》，这篇文章讲合规性测试和端到端的测试。

图片.png

四、合规性测试

合规性测试用于确保配置遵循您为项目定义的策略，并且其应在项目开始时整合到开发周期中。

例如，对于Terraform代码，合规性策略可能有这样的例子：“如果您正在创建 Azure 资源，它必须包含一个标记”。

terraform-compliance工具提供了一个测试框架，我们可以使用它创建策略，然后根据 Terraform 执行plan去运行这些策略。

Note：Terraform Sentinel 也可用于将策略编写为代码，这是一项付费功能，仅适用于 terraform Cloud 或 Enterprise版本。除此之外， Cloudrail也是一个不错的选择。

可参照Azure官方文档的合规性测试-基于Azure的合规性测试。

失败的测试示例

五、端到端测试

端到端测试也被称为“链测试”，是在部署到生产之前验证系统所有的组成部分是否协同工作，它是对整个过程链路的完整测试，端到端测试通常是一个三个步骤，包括：

配置应用于测试环境
然后将运行代码以验证结果
测试环境被重新初始化或关闭（例如取消分配虚拟机）

同样的，Terratest 可用于端到端的测试（以及集成和单元测试）。

可参照Azure官方文档的端到端测试-基于Azure的端到端测试。

示例端到端测试场景

少年，没看够？点击石头的主页，随便点点看看，说不定有惊喜呢？欢迎支持点赞/关注/评论，有你们的支持是我更文最大的动力，多谢啦！

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

你的项目是时候集成RocketMQ了

发表于 2021-10-31

1. RocketMQ介绍

rocketmq是阿里巴巴开源的一款分布式的消息中间件，他源于jms规范但是不遵守jms规范。对于分布式只一点，如果你了用过其他mq并且了解过rocketmq，就知道rocketmq天生就是分布式的，可以说是broker、provider、consumer等各种分布式

2. RocketMQ优点

去除对zk的依赖
支持异步和同步两种方式刷磁盘
单机支持的队列或者topic数量是5w
支持消息重试
支持严格按照一定的顺序发送消息
支持定时发送消息
支持根据消息ID来进行查询
支持根据某个时间点进行消息的回溯
支持对消息服务端的过滤
消费并行度:顺序消费取决于queue数量,乱序消费取决于consumer数量

3. RocketMQ发送消息和消费消息

(1) 创建父工程

pom.xml如下

java复制代码<?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>com.james</groupId>
    <artifactId>rocketmq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <packaging>pom</packaging>
    <modules>
        <module>MQProducer-demo</module>
        <module>MQConsume-demo</module>
    </modules>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>com.james</groupId>
                <artifactId>common_utils</artifactId>
                <version>0.0.3-SNAPSHOT</version>
            </dependency>

            <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.9.2</version>
            </dependency>

            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.22</version>
            </dependency>

        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <!--该配置必须 -->
                    <fork>true</fork>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

        </plugins>

    </build>

</project>

(2) 创建消息生产者

新建工程MQProducer-demo

pom.xml如下所示

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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <artifactId>rocketmq-demo</artifactId>
        <groupId>com.james</groupId>
        <version>1.0-SNAPSHOT</version>
        <relativePath>../pom.xml</relativePath>
    </parent>

    <packaging>jar</packaging>

    <artifactId>MQProducer-demo</artifactId>

    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>com.james</groupId>
            <artifactId>common_utils</artifactId>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>
</project>

修改application.properties文件

properties复制代码server.port=8082

spring.application.name=producer-demo

spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=
spring.redis.database=10

swagger.enable=true

spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8


rocketmq.name-server=localhost:9876
rocketmq.producer.group=2021-11

# 是否开启自动配置
rocketmq.producer.isOnOff=on
# 发送同一类消息设置为同一个group，保证唯一默认不需要设置，rocketmq会使用ip@pid（pid代表jvm名字）作为唯一标识
rocketmq.producer.groupName=${spring.application.name}
# mq的nameserver地址
rocketmq.producer.namesrvAddr=localhost:9876
# 消息最大长度 默认 1024 * 4 (4M)
rocketmq.producer.maxMessageSize = 4096
# 发送消息超时时间，默认 3000
rocketmq.producer.sendMsgTimeOut=3000
# 发送消息失败重试次数，默认2
rocketmq.producer.retryTimesWhenSendFailed=2

新建消息生产者配置类

java复制代码package com.james.mq.producer.config;

import lombok.Data;
import lombok.ToString;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author james
 * @version 1.0
 * @description: 消息生产者配置
 * @date 2021/10/30 11:43
 */
@Data
@ToString
@Configuration
@ConfigurationProperties(prefix = "rocketmq.producer")
public class MQProducerConfigure {


    public static final Logger LOGGER = LoggerFactory.getLogger(MQProducerConfigure.class);

    private String groupName;
    private String namesrvAddr;
    // 消息最大值
    private Integer maxMessageSize;
    // 消息发送超时时间
    private Integer sendMsgTimeOut;
    // 失败重试次数
    private Integer retryTimesWhenSendFailed;

    /**
     * mq 生成者配置
     * @return
     * @throws MQClientException
     */
    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.producer", value = "isOnOff", havingValue = "on")
    public DefaultMQProducer defaultProducer() throws MQClientException {
        LOGGER.info("defaultProducer 正在创建---------------------------------------");
        DefaultMQProducer producer = new DefaultMQProducer(groupName);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setVipChannelEnabled(false);
        producer.setMaxMessageSize(maxMessageSize);
        producer.setSendMsgTimeout(sendMsgTimeOut);
        producer.setRetryTimesWhenSendAsyncFailed(retryTimesWhenSendFailed);
        producer.start();
        LOGGER.info("rocketmq producer server 开启成功----------------------------------");
        return producer;
    }
}

新建消息生产者接口

java复制代码package com.james.mq.producer.controller;

import com.james.common.result.Result;
import com.james.common.utils.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author james
 * @version 1.0
 * @description: 消息生产者
 * @date 2021/10/30 11:12
 */
@RestController
@RequestMapping("/api")
public class ProducerController {

    public static final Logger LOGGER = LoggerFactory.getLogger(ProducerController.class);

    @Autowired
    DefaultMQProducer defaultMQProducer;

    /**
     * 发送简单的MQ消息
     * @param msg
     * @return
     */
    @GetMapping("/send")
    public Result send(String msg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        if (StringUtils.isEmpty(msg)) {
            return Result.OK();
        }
        LOGGER.info("发送MQ消息内容：" + msg);
        Message sendMsg = new Message("HelloTopic", "HelloTag", msg.getBytes());
        // 默认3秒超时
        SendResult sendResult = defaultMQProducer.send(sendMsg);
        LOGGER.info("消息发送响应：" + sendResult.toString());
        return Result.OK(sendResult);
    }

}

测试

生产者控制台发送消息:

截屏2021-10-31 00.20.55

(3) 创建消息消费者

pom.xml同消息生成者模块

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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <artifactId>rocketmq-demo</artifactId>
        <groupId>com.james</groupId>
        <version>1.0-SNAPSHOT</version>
        <relativePath>../pom.xml</relativePath>
    </parent>


    <artifactId>MQConsume-demo</artifactId>

    <packaging>jar</packaging>

    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>com.james</groupId>
            <artifactId>common_utils</artifactId>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

</project>

修改配置文件

properties复制代码spring.application.name=consumers-demo
server.port=8801

spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=
spring.redis.database=10

swagger.enable=true

spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8

# 是否开启自动配置
rocketmq.consumer.isOnOff=on
# 发送同一类消息设置为同一个group，保证唯一默认不需要设置，rocketmq会使用ip@pid（pid代表jvm名字）作为唯一标识
rocketmq.consumer.groupName=${spring.application.name}
# mq的nameserver地址
rocketmq.consumer.namesrvAddr=127.0.0.1:9876
# 消费者订阅的主题topic和tags（*标识订阅该主题下所有的tags），格式: topic~tag1||tag2||tags3;
rocketmq.consumer.topics=TestTopic~TestTag;TestTopic~HelloTag;HelloTopic~HelloTag;MyTopic~*
# 消费者线程数据量
rocketmq.consumer.consumeThreadMin=5
rocketmq.consumer.consumeThreadMax=32
# 设置一次消费信心的条数，默认1
rocketmq.consumer.consumeMessageBatchMaxSize=1

mq消费者配置

java复制代码package com.james.mq.consume.config;

import lombok.Data;
import lombok.ToString;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author james
 * @version 1.0
 * @description: mq消费者配置
 * @date 2021/10/30 17:34
 */
@Data
@ToString
@Configuration
@ConfigurationProperties(prefix = "rocketmq.consumer")
public class MQConsumerConfigure {
    public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfigure.class);

    private String groupName;
    private String namesrvAddr;
    private String topics;
    // 消费者线程数据量
    private Integer consumeThreadMin;
    private Integer consumeThreadMax;
    private Integer consumeMessageBatchMaxSize;

    @Autowired
    private MQConsumeMsgListenerProcessor consumeMsgListenerProcessor;
    /**
     * mq 消费者配置
     * @return
     * @throws MQClientException
     */
    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.consumer", value = "isOnOff", havingValue = "on")
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        LOGGER.info("defaultConsumer 正在创建---------------------------------------");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        // 设置监听
        consumer.registerMessageListener(consumeMsgListenerProcessor);

        /**
         * 设置consumer第一次启动是从队列头部开始还是队列尾部开始
         * 如果不是第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
//        consumer.setMessageModel(MessageModel.CLUSTERING);

        try {
            // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*,
            String[] topicArr = topics.split(";");
            for (String topic : topicArr) {
                String[] tagArr = topic.split("~");
                consumer.subscribe(tagArr[0],"*");
            }
            consumer.start();
            LOGGER.info("consumer 创建成功 groupName={}, topics={}, namesrvAddr={}",groupName,topics,namesrvAddr);
        } catch (MQClientException e) {
            LOGGER.error("consumer 创建失败!");
        }
        return consumer;
    }
}

这个只是初始化操作，实际对消费者对消息处理放在 consumer.registerMessageListener(consumeMsgListenerProcessor); 这个监听类里面了，实际接收消息，处理消息都放在监听类里

java复制代码package com.james.mq.consume.config;

import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author james
 * @version 1.0
 * @description: 消费者监听
 * @date 2021/10/30 17:35
 */
@Component
public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently {
    public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);


    /**
     * 默认msg里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
     * 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
     *
     * @param msgList
     * @param consumeConcurrentlyContext
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if (CollectionUtils.isEmpty(msgList)) {
            LOGGER.info("MQ接收消息为空，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = msgList.get(0);
        LOGGER.info("MQ接收到的消息为：" + messageExt.toString());
        try {
            String topic = messageExt.getTopic();
            String tags = messageExt.getTags();
            String body = new String(messageExt.getBody(), "utf-8");

            LOGGER.info("MQ消息topic={}, tags={}, 消息内容={}", topic, tags, body);
        } catch (Exception e) {
            LOGGER.error("获取MQ消息内容异常{}", e);
        }
        // TODO 处理业务逻辑
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}

启动测试

截屏2021-10-31 00.47.01

如图,收到生产者消息

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

SpringBoot自动配置Quartz

发表于 2021-10-30

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动

SpringBoot2.0版本之后，其中增加了对Quartz框架的支持内容，可以实现通过容器来自动配置Quartz。

依赖信息

springboot2.x版本时，出现了spring-boot-starter-quartz这一起步依赖，其中提供了很多丰富功能。

1.1 原依赖信息

之前引入quartz框架时，必须引入的依赖信息有：

xml复制代码<!--quartz核心包-->
<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>2.3.2</version>
</dependency>
<!--添加Scheduled坐标-->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
</dependency>
<!--Spring tx 坐标-->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-tx</artifactId>
</dependency>

依赖引入之后，项目的maven中的依赖管理信息

1.2 新的依赖

使用新的spring-boot-starter-quartz依赖代替原有三种依赖信息：

xml复制代码<!--springboot2-quartz依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

依赖引入后，项目maven管理的依赖信息

1.3 依赖变化

可以看出，springBoot2.0之后使用一个starter依赖就相当于引入三个依赖信息，spring-boot-starter-quartz本质上与引入三个依赖信息相同。

新的依赖使用

2.1 默认配置可用

使用新的依赖信息后，对于原有的配置方式是完全可行的，如原有的quartz使用流程：

定义任务实现Job，并重写其中的execute()方法，添加执行的任务
配置Quartz配置类，配置类中注入Job、Trigger、Scheduler对象
配置类或者启动类上使用@EnableScheduling注解开启定时任务
项目启动时会自动执行配置的定时任务

在新的依赖支持下项目执行效果完全一致。

2.2 使用自动配置

如果说springboot2.0之后的starter依赖带来的变化，最主要的就是依赖包中封装了quartz的自动配置相关内容。

quartz相关的自动配置类是springboot的autoconfigure自动配置类包中提供的支持，如下

QuartzAutoConfiguration，自动配置类，其中会自动初始化配置调度器类、数据源信息、和数据存储类型等
QuartzProperties，配置文件类，对于quartz的配置信息，使用统一的application.yml/properties管理，jar包中提供了一个QuartzProperties类专门用来获取配置文件中quartz相关的配置信息。
- 该类存在org.springframework.boot.autoconfigure.quartz springboot自动配置包中，获取配置文件中以”spring.quartz”开头的配置
QuartzDataSourceInitializer，初始化数据源操作，直接使用spingboot项目的数据源配置
JobStoreType，定义quartz数据存储类型的枚举类，有MEMORY/JDBC两个值
SchedulerFactoryBeanCustomizer，功能接口，可以通过实现该接口来实现调度器类的自定义配置

使用自动配置类流程

使用starter依赖信息
使用 application.yml/properties 统一管理quartz配置
移除使用@Configuration标注的自动配置类，否则该配置类会代替自动配置

2.3 配置信息：

在application.yml文件中配置数据持久化：

yaml复制代码server:
  port: 8082

spring:
  datasource:
    url: jdbc:mysql://10.35.219.24:3306/test282?autoReconnect=true&useUnicode=true&characterEncoding=utf-8
    username: mysql
    password: Dh2236@db!
    driver-class-name: com.mysql.cj.jdbc.Driver
  quartz:
    #相关属性配置
    properties:
      org:
        quartz:
          scheduler:
            instanceName: clusteredScheduler
            instanceId: AUTO
          jobStore:
            class: org.quartz.impl.jdbcjobstore.JobStoreTX
            driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
            tablePrefix: QRTZ_
            isClustered: true
            clusterCheckinInterval: 10000
            useProperties: false
          threadPool:
            class: org.quartz.simpl.SimpleThreadPool
            threadCount: 15
            threadPriority: 5
            threadsInheritContextClassLoaderOfInitializingThread: true
    #数据库方式
    job-store-type: jdbc

数据库存储方式字段job-store-typ可以取值为jdbc或memory，如果需要设置为memory存储在内存中时，需要则需要更改jobStore.class为内存类型，并移除jonStore下的其他配置信息；否则会启动报错。

总结

SpringBoot的自动配置带来了很大的效率，减去了很多重复的配置、也增加了代码的简洁性。

但是，自动配置往往也会带来一些问题，如自动配置冲突等情况，往往需要手动来选择最终使用的配置对象。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有