作为开发者，我们都喜欢在博客上分享自己踩过的坑、总结出的经验。但很多时候，一篇自认为写得还不错的文章发出去后，除了朋友圈里的几个点赞，几乎无人问津。问题出在哪？内容不够好吗？不一定。更多时候，只是因为"需要它的人找不到它"。
这就是SEO（搜索引擎优化）要解决的问题。一提到SEO，很多人会联想到各种复杂的黑话和技巧，但对于我们技术博客来说，事情可以简单得多。它不是什么黑魔法，而是一系列逻辑清晰的、让搜索引擎（比如Google）更容易理解我们网站内容的步骤。
最近我给自己的博客做了点优化，效果还不错，这里就跟大家分享一下我做了哪些事。
作为后端开发者，我们对 Elasticsearch（后文简称 ES）绝不陌生。它是一个强大的分布式搜索引擎，我们常常用它来做日志分析、全文检索等。但很多时候，我们可能只是停留在"会用"的层面：创建索引、插入数据、用 match 查询。当数据量和服务规模上来后，各种问题便接踵而至："为什么我的查询这么慢？"、"为什么集群CPU突然就满了？"、"分片数到底设多少合适？"
如果你也遇到过这些问题，那么这篇文章就是为你准备的。我们将深入探讨那些能让你的 ES 使用水平提升一个台阶的核心知识：高阶查询语法与分片策略。这不仅仅是"知其然"，更是"知其所以然"的过程，帮助你从根源上理解并解决性能问题。
在实际开发中，树形结构广泛应用于组织和表示层级化数据，如文件目录、菜单导航、组织架构等。Java 8 的流式 API 和函数式编程特性为处理这类问题提供了极大的便利。下面将分享如何使用 Java 8 中的 Function 和 BiConsumer 等新特性，构建树形结构并处理父子节点关系。
在许多业务场景中，我们经常需要根据节点的父子关系来构建一个树形结构。例如，一个简单的交易信息模型可能包含多个交易节点，节点之间有父子关系。通过这种方式，我们可以快速实现层级结构，并对数据进行灵活操作。
我们需要实现一个方法，将一组数据根据父子关系构建成树形结构。在树形结构中，父节点将包含所有子节点。我们使用 Java 8 提供的流式操作来实现这一功能，同时利用函数式接口，如 Function 和 BiConsumer，来提高代码的灵活性和可复用性。
void，我们用它来处理节点之间的父子关系，例如为父节点设置子节点。import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* 功能描述:
* 该类用于构建树形结构，基于给定的数据列表，将父子节点关系组织成层级化的树形结构。
* 通过使用 Java 8 新特性，如 Function 和 BiConsumer，灵活地处理节点间的父子关系。
* -
* 创建人: 季冠臣
* 创建时间: 2024/12/2 15:38
*/
public class TreeStructureTest {
// 模拟的节点类
static class TradeInfo {
private String tradeID; // 当前节点ID
private String parentID; // 父节点ID
private List<TradeInfo> children; // 子节点列表
public TradeInfo(String tradeID, String parentID) {
this.tradeID = tradeID;
this.parentID = parentID;
this.children = new ArrayList<>();
}
// Getter 和 Setter 方法
public String getTradeID() {
return tradeID;
}
public void setTradeID(String tradeID) {
this.tradeID = tradeID;
}
public String getParentID() {
return parentID;
}
public void setParentID(String parentID) {
this.parentID = parentID;
}
public List<TradeInfo> getChildren() {
return children;
}
public void setChildren(List<TradeInfo> children) {
this.children = children;
}
@Override
public String toString() {
return "TradeInfo{" +
"tradeID='" + tradeID + '\'' +
", parentID='" + parentID + '\'' +
", children=" + children +
'}';
}
}
// 构建树形结构
private static <T, R> List<T> buildTree(List<T> dataList, BiConsumer<T, List<T>> setChildF, Function<T, R> getIdF,
Function<T, R> getParentIdF, R rootParentId) {
if (dataList != null && !dataList.isEmpty()) {
setChildCall(dataList, setChildF, getIdF, getParentIdF, rootParentId);
dataList = dataList.stream()
.filter(t -> (rootParentId != null && rootParentId.equals(getParentIdF.apply(t)))
|| (rootParentId == null && getParentIdF.apply(t) == null))
.collect(Collectors.toList());
}
return dataList;
}
// 将子集放入父级
private static <T, R> List<T> setChildCall(List<T> dataList, BiConsumer<T, List<T>> setChildF, Function<T, R> getIdF,
Function<T, R> getParentIdF, R parentId) {
if (dataList == null || dataList.isEmpty()) {
return dataList;
}
List<T> parentChild = dataList.stream()
.filter(t -> parentId == null || (getParentIdF.apply(t) != null && getParentIdF.apply(t).equals(parentId)))
.collect(Collectors.toList());
for (T t : parentChild) {
setChildF.accept(t, setChildCall(dataList, setChildF, getIdF, getParentIdF, getIdF.apply(t)));
}
return parentChild;
}
public static void main(String[] args) {
// 创建示例数据
List<TradeInfo> dataList = Arrays.asList(
new TradeInfo("1", null),
new TradeInfo("2", "1"),
new TradeInfo("3", "1"),
new TradeInfo("4", "2"),
new TradeInfo("5", "2")
);
// 定义如何获取ID和父ID
Function<TradeInfo, String> getIdF = TradeInfo::getTradeID;
Function<TradeInfo, String> getParentIdF = TradeInfo::getParentID;
// 定义如何设置子节点
BiConsumer<TradeInfo, List<TradeInfo>> setChildF = TradeInfo::setChildren;
// 构建树结构
List<TradeInfo> tree = buildTree(dataList, setChildF, getIdF, getParentIdF, null);
// 输出结果
System.out.println("构建后的树形结构:");
tree.forEach(System.out::println);
}
}
tradeID 和 parentID，并通过 children 列表存储子节点。BiConsumer 将子节点设置到父节点，构建树形结构。Function 接口获取节点的 ID 和父 ID，通过 BiConsumer 设置子节点，实现了灵活的节点关系操作。运行以上代码后，输出的树形结构将展示如下：
构建后的树形结构:
TradeInfo{tradeID='1', parentID='null', children=[TradeInfo{tradeID='2', parentID='1', children=[TradeInfo{tradeID='4', parentID='2', children=[]}, TradeInfo{tradeID='5', parentID='2', children=[]}]}, TradeInfo{tradeID='3', parentID='1', children=[]}]}
如上所示，我们成功地将交易信息列表组织成了树形结构。
通过 Java 8 的函数式编程特性，我们可以更加简洁、灵活地处理复杂的父子节点关系。这种方法不仅适用于树形结构的构建，还能够广泛应用于其他需要层级关系的数据处理场景。例如，在处理文件目录、组织架构、菜单层级等问题时，采用类似的方法将大大提高代码的可读性和可维护性。
当然，在实际应用中，树形结构可能会更加复杂，涉及更多的业务逻辑。你可以根据自己的需求进行扩展和调整。希望本文能为你的开发工作提供一些启发！🚀
本文介绍如何在Nacos中使用国密SM4算法对敏感信息进行加密，并在Spring Boot项目中解密这些配置项。使用Jasypt来集成加密库，并编写一个自定义的加密器，以实现配置项的加密和解密功能。
<dependency>
<groupId>com.github.ulisesbocchio</groupId>
<artifactId>jasypt-spring-boot</artifactId>
<version>3.0.4</version>
</dependency>
注：需要引用高版本的3.0.4，避免影响系统性能，参考 https://www.jianshu.com/p/c02276987d46
创建国密SM4加密工具类，它将负责实际的加密和解密操作。已有工具所在目录 com.space.jiguanchen.utils.Sm4Utils ：
package com.space.jiguanchen.utils;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.pqc.math.linearalgebra.ByteUtils;
import javax.crypto.Cipher;
import javax.crypto.KeyGenerator;
import javax.crypto.spec.SecretKeySpec;
import java.security.Key;
import java.security.SecureRandom;
import java.security.Security;
import java.util.Arrays;
/**
* @author jiguanchen
*/
public class Sm4Utils {
public static final String ENCODING = "UTF-8";
public static final String ALGORIGTHM_NAME = "SM4";
public static final String ALGORITHM_NAME_ECB_PADDING = "SM4/ECB/PKCS7Padding";
public static final int DEFAULT_KEY_SIZE = 128;
public static final String key="dhjashdasa26kl223jkwqe2ejwejwq";
public Sm4Utils() {
}
static {
Security.addProvider(new BouncyCastleProvider());
}
/**
* @Description:生成ecb暗号
*/
private static Cipher generateEcbCipher(String algorithmName, int mode, byte[] key) throws Exception {
Cipher cipher = Cipher.getInstance(algorithmName,BouncyCastleProvider.PROVIDER_NAME);
Key sm4Key = new SecretKeySpec(key, ALGORIGTHM_NAME);
cipher.init(mode, sm4Key);
return cipher;
}
/**
* @Description:自动生成密钥
*/
public static byte[] generateKey() throws Exception {
return generateKey(DEFAULT_KEY_SIZE);
}
public static byte[] generateKey(int keySize) throws Exception {
KeyGenerator kg = KeyGenerator.getInstance(ALGORIGTHM_NAME, BouncyCastleProvider.PROVIDER_NAME);
kg.init(keySize, new SecureRandom());
return kg.generateKey().getEncoded();
}
/**
* @Description:加密
*/
public static String encryptEcb(String hexKey, String paramStr, String charset) throws Exception {
String cipherText = "";
if (null != paramStr && !"".equals(paramStr)) {
byte[] keyData = ByteUtils.fromHexString(hexKey);
charset = charset.trim();
if (charset.length() <= 0) {
charset = ENCODING;
}
byte[] srcData = paramStr.getBytes(charset);
byte[] cipherArray = encrypt_Ecb_Padding(keyData, srcData);
cipherText = ByteUtils.toHexString(cipherArray);
}
return cipherText;
}
/**
* @Description:加密模式之ecb
*/
public static byte[] encrypt_Ecb_Padding(byte[] key, byte[] data) throws Exception {
Cipher cipher = generateEcbCipher(ALGORITHM_NAME_ECB_PADDING, Cipher.ENCRYPT_MODE, key);
byte[] bs = cipher.doFinal(data);
return bs;
}
/**
* @Description:sm4解密
*/
public static String decryptEcb(String hexKey, String cipherText, String charset) throws Exception {
String decryptStr = "";
byte[] keyData = ByteUtils.fromHexString(hexKey);
byte[] cipherData = ByteUtils.fromHexString(cipherText);
byte[] srcData = decrypt_Ecb_Padding(keyData, cipherData);
charset = charset.trim();
if (charset.length() <= 0) {
charset = ENCODING;
}
decryptStr = new String(srcData, charset);
return decryptStr;
}
/**
* @Description:解密
*/
public static byte[] decrypt_Ecb_Padding(byte[] key, byte[] cipherText) throws Exception {
Cipher cipher = generateEcbCipher(ALGORITHM_NAME_ECB_PADDING, Cipher.DECRYPT_MODE, key);
return cipher.doFinal(cipherText);
}
/**
* @Description:密码校验
*/
public static boolean verifyEcb(String hexKey,String cipherText,String paramStr) throws Exception {
boolean flag = false;
byte[] keyData = ByteUtils.fromHexString(hexKey);
byte[] cipherData = ByteUtils.fromHexString(cipherText);
byte[] decryptData = decrypt_Ecb_Padding(keyData,cipherData);
byte[] srcData = paramStr.getBytes(ENCODING);
flag = Arrays.equals(decryptData,srcData);
return flag;
}
/**
* @Description:测试类
*/
public static void main(String[] args) {
try {
// 自定义的32位16进制密钥
String key = "dhjashdasa26kl223jkwqe2ejwejwq";
String url = Sm4Utils.encryptEcb(key, "jdbc:mysql://127.0.0.1:3306/demo?serverTimezone=GMT%2B8",ENCODING);
String username = Sm4Utils.encryptEcb(key, "root",ENCODING);
String password = Sm4Utils.encryptEcb(key, "root",ENCODING);
System.out.println(url);
System.out.println(username);
System.out.println(password);
} catch (Exception e) {
e.printStackTrace();
}
}
}
Sm4Utils类包含了国密SM4算法的加密和解密方法。
为了让Nacos使用我们自己的加密工具进行配置解密，我们需要创建一个自定义的Jasypt加密器。在项目中新建一个名为JasyptStringEncryptor的类，并让它实现jasypt提供的StringEncryptor接口，如下所示：
package com.space.jiguanchen.auth.config;
import com.space.jiguanchen.utils.Sm4Utils;
import lombok.extern.slf4j.Slf4j;
import org.jasypt.encryption.StringEncryptor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
/**
* @author jiguanchen
*/
@Component
@Slf4j
public class JasyptStringEncryptor implements StringEncryptor {
@Value("${my.jasyptEncryptor.enable:false}")
private boolean enable;
@Override
public String encrypt(String message) {
final String exMsg = String.format("encrypt method is not implemented, original message: '%s'.", message);
log.error(exMsg);
throw new UnsupportedOperationException(exMsg);
}
@Override
public String decrypt(String message) {
String decryptedMessage = null;
try {
if (enable){
decryptedMessage = Sm4Utils.decryptEcb(Sm4Utils.key,message,Sm4Utils.ENCODING);
}else {
decryptedMessage=message;
}
log.debug("---------------------------------------解密了: {} -----------------------------------",decryptedMessage);
} catch (Exception e) {
log.error("decrypt failed:"+message+"-"+ e);
throw new RuntimeException(e);
}
return decryptedMessage;
}
}
在JasyptStringEncryptor中，我们注入了一个名为enable的属性，该属性用于控制加密功能是否启用。当enable为true时，我们使用我们自己的国密SM4工具对配置信息进行解密；当enable为false时，我们直接返回原始配置信息，即不进行解密操作。
注：如果自定义类名不是JasyptStringEncryptor，则需要在nacos上面添 加配置：
jasypt:
encryptor:
## 实现jasypt加加密解密的类
bean: JasyptStringEncryptorDemo
最后，我们需要在Nacos配置文件中指定使用自定义的加密器。在bootstrap.properties或bootstrap.yml文件中添加以下配置：
# 启用自定义加密器
my:
jasyptEncryptor:
enable: false
这里的my.jasyptEncryptor.enable属性的值设置为true，表示启用我们自己编写的加密逻辑，即使用国密SM4进行解密。如果要禁用加密功能，将其设置为false。
完成了以上步骤后，Nacos会在启动时自动加载并使用我们的自定义加密器，从而实现数据库配置的解密操作。
使用国密SM4加密工具类Sm4Utils来加密敏感配置项，例如数据库连接的url、username和password。在Sm4Utils的main方法中进行加密操作，得到加密后的配置项：
public static void main(String[] args) {
try {
// 自定义的32位16进制密钥
String key = "dhjashdasa26kl223jkwqe2ejwejwq";
String url = Sm4Utils.encryptEcb(key, "jdbc:mysql://127.0.0.1:3306/demo?serverTimezone=GMT%2B8",ENCODING);
String username = Sm4Utils.encryptEcb(key, "root",ENCODING);
String password = Sm4Utils.encryptEcb(key, "root",ENCODING);
System.out.println(url);
System.out.println(username);
System.out.println(password);
} catch (Exception e) {
e.printStackTrace();
}
}
得到加密后的密文配置项，将它们作为值存放到Nacos中的配置项中：
url=ENC(opkdai32klsamnxa97b678f092c50e042f756371ef644de9dc4aa6318afa8a304a20b327ce84eb573ade0fb7336be9827cec08cd149f0c5380af4926f5219b7829edde1b4026d4229d30a86702851948164bc23af18a4797bae23f7833c2239ab3fd654dc234be37b2b3f31cbeae9e09eaf519a04a623d3c5795d23fa8fa3ce1c59eaf87c0af
username=ENC(dasjkdljasdjad23342mkljjlk)
password=ENC(0298hsdajd328783jdsakjd212)
重启服务，在Nacos启动时会被正确地解密并加载到Spring Boot项目中。
背景：RocketMQ是阿里团队研发并开源的消息中间件，它延续了kafka的很多优点，例如高性能，为了研究RocketMQ究竟是如何做到高性能的呢？零拷贝技术就是一个重要的原因。
高效原因：
CommitLog顺序写, 存储了MessagBody、message key、tag等信息
ConsumeQueue随机读 + 操作系统的PageCache + 零拷贝技术ZeroCopy
零拷贝技术
read(file, tmp_buf, len);
write(socket, tmp_buf, len);
例子：将一个File读取并发送出去（Linux有两个上下文，内核态，用户态）
ZeroCopy：
原始的拷贝技术：
零拷贝技术：
背景：我们知道在Ngnix的conf配置文件之后,不能kill掉matser重新启动Ngnix服务器，而是需要执行 nginx -s reload命令，在 不停止服务始终在处理新的请求的同时把 nginx 的配置文件平滑的把旧的 nginx.conf 配置更新为新的 nginx.conf 配置。那么这是为什么呢？
这样一个功能对于 nginx 非常有必要，但是有时候我们会发现在执行 nginx -s reload 命令后，worker 子进程的数量会变多了，这是因为老的配置运行的 worker 进程长时间没有退出，当使用 stream 做四层反向代理的时候，可能这种场景会更多。
那么下面我们通过分析 nginx 的 reload 流程，来探究下 nginx 到底做了些什么？所谓优雅的退出和立即退出有什么区别？
第一步在修改好 nginx 的配置文件 nginx.conf 后，向 master 进程发送 HUP 信号，这实际上和我们在命令行执行 nginx -s reload 命令效果是一样的。
那么 master 进程在收到 HUP 信号以后，会在第二步检查我们的配置文件语法是否正确，也就是说我们并不一定非要在 nginx -s reload 前执行 nginx -t 检验下语法是否正确，因为在第二步 nginx 的 master 进程一定会执行这个步骤。
在 nginx 的配置语法全部正确以后，master 进程会打开新的监听端口，为什么要在 master 进程中打开新的监听端口？因为我们可能在 nginx.conf 中会引入新的例如 443 或者之前我们没有打开的的监听端口，而所有 worker 进程是 master 进程 的子进程，子进程会继承父进程所有已经打开的端口，这是 linux 操作系统定义的，所以第三步，我们 master 进程打开了可能引入的新的监听端口。
接下来 mster 进程会用新的 nginx.conf 配置文件来启动新的 worker 子进程，那么老的 worker 子进程会怎么样呢？
我们会在第五步在启动新的 worker 子进程以后，由 master 进程再向老 worker 子进程发送 QUIT 信号，QUIT 信号和 TERM，INT 信号是不一样的，QUIT 信号是请优雅地关闭子进程，这时候需要关注顺序，因为 nginx 需要保证平滑，所以要先启动新的 worker 子进程，再向老的 worker 子进程发送 QUIT 信号。
那么老的 master 子进程收到 QUIT 信号后，首先关闭监听句柄，也就是说这个时候新的连接只会到新的 worker 子进程，所以虽然他们之间有时间差，但是时间是非常快速的，那么关闭监听句柄后，处理完当前连接后就结束进程。
下面看 reload 不停机载入新配置的图示。
master 进程上原先有四个绿色的 worker 子进程，它们使用了老的配置，当我们更改了 nginx.conf 配置文件后，向 master 发送 SIGHUP 信号或者执行 reload 命令， 然后 master 会用新的配置文件启动四个新的 黄色 worker 子进程，此时是四个老的绿色 worker 子进程和四个新的黄色的 worker 子进程是并存的。那么老的 worker 子进程在正常的情况下会在处理已经建立好的连接上的请求之后关闭这个连接，哪怕这个连接是 keeplive 请求也会正常关闭。
但是异常情况，如果有一些请求出现问题，客户端长时间无法处理，那么就会导致这个请求长时间停留在这个 worker 子进程当中，那么这个 worker 子进程会长时间存在，因为新的连接已经跑在黄色的 worker 子进程中，所以影响并不会很大，唯一会影响的就是绿色的 worker 子进程会长时间存在，但也只影响已存在的连接，不会影响新的连接。
我们有什么办法处理呢？在新版本中提供了一个新的配置 worker_shutdown_timeout，也就是说最长等待多长时间，这样 master 进程启动新的黄色 worker 进程之后，如果老的 worker 进程一直没有退出，时间到了之后会强制把老的 worker 进程退出掉。
本文主要讲解了 Nginx 平滑升级新的配置文件的流程，在我们了解了优雅关闭 worker 子进程和启动新配置的 worker 子进程流程间的关系后，我们可以更好地处理罕见的异常场景。
在Java开发中，处理Excel文件是一个常见的需求。EasyExcel作为一个流行的Excel操作库，提供了方便而高效的API来读写Excel文件。然而，有时会遇到数据第一行被误读为表头的问题，特别是在Excel文件的第一行不是标准表头而是实际数据时，这一问题显得尤为突出。本文将详细讨论这一问题的根本原因，并提供一种有效的解决方案。
EasyExcel是阿里巴巴开源的一款Java操作Excel的工具库，它提供了强大的功能，支持大数据量的读写操作，并且提供了丰富的样式和格式处理功能，适用于各种场景下的Excel文件处理需求。你可以访问 EasyExcel 的官方 GitHub 页面获取更多的资料和下载： https://github.com/alibaba/easyexcel
在使用EasyExcel读取Excel文件时，经常会出现第一行数据 被错误地识别为表头的情况。这一问题的根本原因在于EasyExcel在某些情况下无法正确识别Excel文件中数据行和表头行的区分，特别是当Excel文件结构比较复杂或者存在特定格式时，EasyExcel的默认解析逻辑可能会出现偏差。
为了解决数据第一行获取问题，我们可以采取以下步骤来调整和优化EasyExcel的读取操作，确保能够正确获取实际数据行而非表头行：
在读取Excel文件时，手动指定数据的起始行，而不依赖EasyExcel的自动识别。这可以通过设置headRowNumber来实现，明确告知EasyExcel从第几行开始读取数据。
ExcelReaderBuilder readBuilder = EasyExcel.read(inputStream, ExcelData.class, new ExcelDataListener())
.sheet().headRowNumber(2) // 指定从第3行开始读取数据
.doRead();
在上述代码中，通过headRowNumber(2)指定从第3行开始读取数据，避免将第一行误读为表头。
数据处理逻辑中排除表头行
在实际数据处理逻辑中，可以通过逻辑判断排除表头行，确保只处理实际的数据行。例如，在invoke方法中可以添加逻辑判断：
@Override
public void invoke(ExcelData data, AnalysisContext context) {
if (context.readRowHolder().getRowIndex() > 0) { // 跳过表头行
// 处理实际数据逻辑
}
}
通过context.readRowHolder().getRowIndex() > 0判断当前行索引大于0时才处理数据，跳过表头行的处理。
使用后处理器进行二次处理
EasyExcel提供了后处理器（Handler）机制，在数据读取完成后可以进行二次处理。可以在doAfterAllAnalysed方法中对数据进行进一步处理或过滤，确保最终数据的准确性和完整性。
@Override
public void doAfterAllAnalysed(AnalysisContext context) {
// 数据处理完成后的逻辑
processData(dataList);
}
在上述代码中，可以在doAfterAllAnalysed方法中调用processData方法，对数据进行进一步的处理或者存储操作。
示例代码
以下是一个完整的示例代码，展示了如何使用EasyExcel读取Excel文件并处理数据，同时避免数据第一行被误读为表头的问题：
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.event.AnalysisEventListener;
import com.alibaba.excel.read.builder.ExcelReaderBuilder;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
public class ExcelReaderExample {
public static void main(String[] args) {
String fileName = "path/to/your/excel/file.xlsx";
InputStream inputStream = null;
try {
inputStream = new FileInputStream(fileName);
ExcelReaderBuilder readBuilder = EasyExcel.read(inputStream, ExcelData.class, new ExcelDataListener());
readBuilder.sheet().headRowNumber(2); // 指定从第3行开始读取数据
readBuilder.doRead();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (inputStream != null) {
try {
inputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
public static class ExcelDataListener extends AnalysisEventListener<ExcelData> {
private List<ExcelData> dataList = new ArrayList<>();
@Override
public void invoke(ExcelData data, AnalysisContext context) {
if (context.readRowHolder().getRowIndex() > 0) { // 跳过表头行
// 处理实际数据逻辑
dataList.add(data);
}
}
@Override
public void doAfterAllAnalysed(AnalysisContext context) {
// 数据处理完成后的逻辑
processData(dataList);
}
private void processData(List<ExcelData> dataList) {
// 处理数据的具体逻辑
for (ExcelData data : dataList) {
System.out.println(data.toString());
}
}
}
public static class ExcelData {
// Excel中的数据字段对应的Java属性
private String column1;
private String column2;
// 省略getter和setter方法
}
}
@Override
@SneakyThrows
public byte[] exportExcel(List<String> sheet1Ids, List<String> sheet2Ids, List<String> sheet3Ids, String startDate, String endDate) {
byte[] result;
// 读取模版
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
InputStream inputStream = templateFileService.getTemplateInputStreamByPath(dataComparisonFileResource.getPath())) {
ExcelWriter excelWriter = EasyExcel.write(outputStream)
.withTemplate(inputStream)
.autoCloseStream(true)
.registerWriteHandler(new CustomSheet1CellWriteHandler())
.build();
// 写入表1 数据
List<LfDispatchDataComparisonGridOperationVo> sheet1Data = getGridOperationData(sheet1Ids, startDate, endDate);
this.writerSheet1(excelWriter, sheet1Data);
// 写入表2 数据
List<LfDispatchDataComparisonGwEnergyVo> sheet2Data = getGwEnergyData(sheet2Ids, startDate, endDate);
this.writerSheet2(excelWriter, sheet2Data);
// 写入表3 数据
List<LfDispatchDataComparisonNewEnergyOperationVo> sheet3Data = getNewEnergyOperationData(sheet3Ids, startDate, endDate);
this.writerSheet3(excelWriter, sheet3Data);
// 生成excel
excelWriter.finish();
// 确保所有内容都写入输出流中
result = outputStream.toByteArray();
}
return result;
}
/**
* 自定义sheet1的cell样式 用于标记 负荷占比、电量占比 大于1的单元格
*/
public static class CustomSheet1CellWriteHandler implements WorkbookWriteHandler, CellWriteHandler {
/**
* 处理第一行数据 (easyExcel工具的bug 会将第一行数据读在表头)
*
* @param context 上下文
*/
@Override
public void afterCellDispose(CellWriteHandlerContext context) {
if (context.getRowIndex() == 2 && (context.getColumnIndex() == 14 || context.getColumnIndex() == 15)) {
// 处理第第一个sheet
Sheet sheet = context.getRow().getSheet().getWorkbook().getSheetAt(0);
Row row = sheet.getRow(2);
Cell cell14 = row.getCell(14);
Cell cell15 = row.getCell(15);
// 应用红色样式
applyRedStyleToCells(cell14, cell15);
}
}
/**
* 处理除了第一行以外的其他数据行
*
* @param context 上下文
*/
@Override
public void afterWorkbookDispose(WorkbookWriteHandlerContext context) {
Workbook workbook = context.getWriteWorkbookHolder().getWorkbook();
// 只处理 sheet1
Sheet sheet = workbook.getSheetAt(0);
applyRedStyleToSheet(sheet);
}
/**
* 按列标记红色
*
* @param sheet 工作表
*/
private void applyRedStyleToSheet(Sheet sheet) {
for (Row row : sheet) {
applyRedStyleToCells(row.getCell(14), row.getCell(15));
}
}
/**
* 对指定的单元格应用红色字体样式
*
* @param cells 需要应用红色样式的单元格
*/
private void applyRedStyleToCells(Cell... cells) {
for (Cell cell : cells) {
if (cell != null) {
Workbook workbook = cell.getSheet().getWorkbook();
CellStyle redStyle = workbook.createCellStyle();
Font redFont = workbook.createFont();
redFont.setColor(IndexedColors.RED.getIndex());
redStyle.setFont(redFont);
cell.setCellStyle(redStyle);
}
}
}
}
我总结：建议不再使用EasyExcel工具。
在分布式系统中，Redis作为消息队列使用时，经常会面临连接中断的风险，这可能由于网络波动、Redis服务崩溃或其他外部因素导致。为了保证系统的高可用性和消息的可靠性，我们需要实现一个自动重试机制，在Redis连接断开时能够自动重连，并保证消息的消费不中断或丢失。本文将重点介绍如何在Redis队列的消费者端实现连接断开后的重连机制。
在我们的架构中，消费者从Redis队列中拉取并处理消息，通常使用Jedis作为客户端与Redis交互。然而，在生产环境中，由于网络延迟、Redis服务重启或Redis连接池被耗尽等原因，Redis连接可能会出现断开。当连接断开时，消费者无法继续拉取消息，导致消息处理中断。
为了解决这个问题，我们需要设计一个重连机制：
消费者类通过定时检查连接状态，并在Redis连接出现问题时尝试重连。我们通过引入一个标志位isDown来表示当前消费者是否处于断线状态。在连接恢复后，消费者能够自动恢复消息消费。
package com.example.redisQueue.consumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Redis消息队列消费者断连重试机制实现
*/
@Slf4j
@Component
public class RedisQueueConsumerListenerRunner implements Runnable {
@Autowired
private MyAbstractRedisQueueConsumer<?> consumer;
/**
* 是否宕机标志
* 通过static保证全局共享该标志
*/
public static AtomicBoolean isDown = new AtomicBoolean(false);
@Override
public void run() {
log.info("消费者启动，开始监听Redis队列...");
// 启动消息消费线程
Executors.newSingleThreadExecutor().submit(() -> {
try {
consumer.consume();
} catch (Exception e) {
log.error("消费过程中发生异常", e);
isDown.set(true); // 设置断开标志
}
});
// 启动断线重试机制
startRetryMechanism();
}
/**
* 启动断连重试机制
*/
public void startRetryMechanism() {
new Thread(() -> {
while (true) {
try {
Thread.sleep(30 * 1000L); // 每30秒检查一次连接状态
if (isDown.get()) {
log.warn("Redis连接断开，正在尝试重连...");
Thread.sleep(3000L); // 延迟3秒后重试
// 执行重连操作
consumer.reconnect();
isDown.set(false);
log.info("Redis连接恢复，重试成功");
}
} catch (InterruptedException e) {
log.error("重试线程被中断", e);
Thread.currentThread().interrupt();
}
}
}).start();
}
}
消费者类需要提供一个reconnect方法，用于在连接断开时重新初始化连接，恢复消费任务。
package com.example.redisQueue.consumer;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Redis消息队列消费者基类
*/
public abstract class MyAbstractRedisQueueConsumer<T> {
private Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private JedisPool jedisPool;
private String queueName;
public MyAbstractRedisQueueConsumer(String queueName) {
this.queueName = queueName;
}
/**
* 重连方法
* 用于连接断开后恢复连接
*/
public void reconnect() {
try (Jedis jedis = jedisPool.getResource()) {
if (jedis.isConnected()) {
logger.info("Redis连接已恢复，消费者已重新连接到队列");
} else {
logger.error("无法恢复Redis连接，正在尝试重新初始化");
// 这里可以进一步增加对Redis连接池或Jedis的重初始化逻辑
}
} catch (Exception e) {
logger.error("Redis重连失败", e);
}
}
/**
* 消费消息的抽象方法
*/
public abstract void doConsume(T message);
/**
* 消费消息
*/
public void consume() throws Exception {
try (Jedis jedis = jedisPool.getResource()) {
while (true) {
// 从Redis队列中拉取消息
List<String> message = jedis.blpop(0, queueName);
if (message != null && message.size() > 1) {
doConsume(message.get(1)); // 处理消息
}
}
} catch (Exception e) {
logger.error("消费过程中发生异常", e);
throw e;
}
}
}
通过Executors.newSingleThreadExecutor()我们为消费者启动了独立的线程来消费消息，并在消费者检测到连接断开时启动重试机制。
Executors.newSingleThreadExecutor().submit(() -> {
try {
consumer.consume(); // 启动消费
} catch (Exception e) {
log.error("消费过程中发生异常", e);
isDown.set(true); // 设置断开标志
}
});
在重试机制中，我们每30秒检查一次isDown标志，如果发现连接已经断开，就延迟3秒后尝试重新连接。
if (isDown.get()) {
log.warn("Redis连接断开，正在尝试重连...");
Thread.sleep(3000L); // 延迟3秒后重试
consumer.reconnect(); // 执行重连操作
isDown.set(false); // 重连后清除断开标志
log.info("Redis连接恢复，重试成功");
}
static AtomicBoolean与替代方案虽然AtomicBoolean是一种简单且有效的实现方式，但我们也可以考虑其他方案，以下是几种常见的替代方案对比。
| 技术方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
static AtomicBoolean | - 简单易用，线程安全<br>- 适用于单一线程或少数线程的全局状态控制 | - 随着系统复杂度增加，可能变得不够灵活<br>- 共享全局状态可能导致问题 | 适合小型系统或少量线程的全局状态管理 |
CountDownLatch | - 可以灵活地控制多个线程同步<br>- 适合等待多个线程完成操作 | - 一次性触发，无法重复使用 | 适合一次性控制多线程完成后进行重试操作 |
ScheduledExecutorService | - 定时检查，易于控制<br>- 支持定期任务 | - 适合周期性任务，不适用于即时响应需求 | 适合定期检查重试的场景 |
EventBus（消息机制） | - 松耦合，适合分布式环境<br>- 可以广播事件通知其他模块 | - 增加了系统复杂性，依赖额外的库 | 适合分布式或微服务架构的消息广播机制 |
RetryTemplate | - 灵活控制重试策略，如重试次数、间隔时间等<br>- 简化重试逻辑 | - 引入外部依赖<br>- 配置和使用稍显复杂 | 适合需要复杂重试逻辑的场景 |
static AtomicBoolean：适合小型系统，能够简单地控制全局状态。缺点是随着系统的复杂度增加，可能变得不够灵活，且状态管理可能存在问题。CountDownLatch：适合等待多个线程完成任务后进行统一操作，但无法多次触发，因此不适合需要持续重连的场景。ScheduledExecutorService：适合定期检查连接状态并进行重试的场景。简单易用，但适合周期性任务，不适合即时响应。EventBus：适合分布式系统中的事件驱动场景，可以实现松耦合的异步通知机制，但会引入额外的复杂性。
RetryTemplate：适合需要复杂重试策略的场景，能够灵活配置重试次数和间隔时间，但相对复杂。根据实际需求选择合适的方案，可以有效地提升系统的稳定性和可用性。