作者：京东工业 宛煜昕

概要
： 1. 调⽤链系统概述； 2. 调⽤链系统的演进； 3. 调⽤链的底层实现逻辑； 4. Span内容组成。

⼀、分布式调⽤链系统概述

客户打电话给客服说：“优惠券使⽤不了”。 -客服告诉运营⼈员 --运营打电话给技术负责⼈ ---技术负责⼈通知会员系统开发⼈员 ----会员找到营销系统开发⼈员 -----营销系统开发⼈员找到DBA ------DBA找到运维⼈员 -------运维⼈员找到机房负责⼈ --------机房负责⼈找到⼀只⽼⿏ ，因为就是它把⽹线咬断了。

分布式架构所带来的问题

定位⼀个问题怎么会如此复杂？竟然动⽤了公司⼀半以上的职能部⻔。但其实这只是当我系统变成分布式之后，当我们把服务进⾏细粒度的拆份之后的⼀⼩部分问题，更多问题在哪⾥？⽐如： 1. 开发成本增加。 2. 测试成本增加。 3. 产品迭代周期将变⻓。 4. 运维成本增加。

问题产⽣原因

在传统制造业，分⼯越精细，专业化程度越⾼，产能就越⾼。⽐如⼀台汽⻋平均将近3万个零部件，来⾃全球各个供应商，最后再由汽⻋⼚商统⼀拼装检测出⼚。不仅⼤件是精细分⼯完成，⼩件也是如此，在浙江温州 有⼀个打⽕机村，⼀个⼩⼩的打⽕机⽣产，是由20多个⼚家协作完成，有的做打⽕机燃料有的做点⽕器。

反观软件⾏业，这种精细分⼯很难实现， 你⻅过哪家某个系统是由⼗⼏家企业协作完成的么？你觉得淘宝的电商系统可以让⽇本⼈去开发 购物⻋模块、让法国⼈实现评论模块、让印度⼈去实现下单功能、美国⼈实现商品模块，最后在由中国⼈拼装整合？究期原因再于三个字：“标准化”，刚说的汽⻋3万个零件，每个都有其标准化规格，所以才能够顺利的拼装成品，但软件组成很难标准，就连开发个接⼝都没有指定标准，就连⼀个规范都难于推⾏。没有标准化，不能分⼯协作，那怎么实现软件的⼤规模⽣产呢？就是⽤更多的⼈，更多⼯作时⻓去冲抵。软件开发就此成为⼀个劳动密集型产业，新⽣代信息化农⺠⼯群体诞⽣。这对企业⽽⾔是不利的，因为它要为信息化付出更多的成本。所以相应管理办法与开发⼯具都要升级，管理办法是类似于敏捿开发、⼯程师⽂化建设、开发形为准则。另外⼀个就是⼯具：⾃动化构建、⾃动化部署、⾃动化运维、⾃动化扩容等、线上链路监控等等。

分布式链路监控的作用

1. 定位线上问题； 2. 分极性能问题； 3. 降纸软件复杂度； 4. 提供决策数据⽀持。

⼆、调用链系统的演进

⼀般我们认为链路监控产品是从 2010 年 Google 发表名为 《Dapper⼤规模分布式系统的跟踪系统》论⽂开始流⾏起来的。之后出现的很多开源或者闭源的产品都是以 Dapper 为理论基础。下表列出已知的链路监控系统。

链路监控系统列表

淘宝鹰眼
鹰眼界面

鹰眼架构

Google Dapper

Dapper 界⾯

Dapper架构图

开源链路监控

三、调用链系统的底层实现逻辑

调用链系统的本质

⼀张⽹⻚，要经历怎样的过程，才能抵达⽤户⾯前?

⽹络传输层

负载均衡层

系统服务层

调用链基本元素

1. 事件：请求处理过程当中的具体动作。

2. 节点：请求所经过的系统节点，即事件的空间属性。

3. 时间：事件的开始和结束时间。

4. 关系：事件与上⼀个事件关系。

调⽤链系统本质上就是⽤来回答这⼏问题:

1. 什么时间？

2. 在什么节点上？

3. 发⽣了什么事情？

4. 这个事情由谁发起？

事件捕捉

1. 硬编码埋点捕捉

2. AOP埋点捕捉

3. 公开组件埋点捕捉

4. 字节码插桩捕捉

事件串联

事件串联的⽬的：

1. 所有事件都关联到同⼀个调⽤

2. 各个事件之间层级关系

为了到达这两个⽬的地，⼏乎所有的调⽤链系统都会有以下两个属性：

traceID：在整个系统中唯⼀，该值相同的事件表示同⼀次调⽤。

spanD：在⼀次调⽤中唯⼀、并展出事件的层级关系

1、怎么⽣成TraceID

2、怎么传递参数

3、怎么并发情况下不允响传递的结果

串联的过程：

1. 由跟踪的起点⽣成⼀个TraceId， ⼀直传递⾄所有节点，并保存在事件属性值当中。

2. 由跟踪的起点⽣成初始SpanId，每捕捉⼀个事件ID加1，每传递⼀次，层级加1。

trackId与SpanId 的传递

SpanId ⾃增⽣成⽅式

我们的埋点是埋在具体某个实现⽅法类，当多线程调⽤该⽅法时如何保证⾃增正确性？

解决办法是每个跟踪请求创建⼀个互相独⽴的会话，SpanId的⾃增都基于该会话实现。通常会话对象的存储基于ThreadLocal实现。

事件的开始与结束

我们知道⼀个事件是⼀个时间段内系统执⾏的若⼲动作，所以对于事件捕捉必须包含开启监听和结束监听两个动作？如果⼀个事件在⼀个⽅法内完成的，这个问题是⽐较好解决的，我们只要在⽅法的开始创建⼀个Event对象，在⽅法结束时调⽤该对像的close ⽅法即可。

但如果⼀个事件的开始和结束触发分布在多个对象或⽅法当中，情况就会变得异常复杂。

⽐如⼀个JDBC执⾏事件，应该是在构建 Statement 时开始，在Statement 关闭时结束。怎样把这两个触发动作对应到同⼀个事件当中去呢（即传递Event对象）？在这⾥的解决办法是对返回结果进⾏动态代理，把Event放置到代理对象的属性当中，以达到付递的⽬标。当这个⽅法只是适应JDBC这⼀个场景，其它场景需要重新设计Event 传递路径，⽬前还没有通⽤的解决办法。

上传

上传有两种⽅式

1. 基于RPC直接上传

2. 打印⽇志，然后在基于Flume或Logstash采集上传。

第⼀种相对简单，直接把数据发送服务进⾏持久化，但如果系统流量较⼤的情况下，会影响系统本身的性能，造成压力。

第⼆种相对复杂，但可以应对⼤流量，通常情况下会采⽤第⼆种解决办法。

四、Span内容组成

Span基本内容

在调⽤链中⼀个Span，即代表⼀个时间跨度下的行为动作，它可以是在⼀个系统内的时间跨度，也可能是跨多个服务系统的。下图即是Dapper中关于Span的描述。

通常情况下⼀个Span组成包括： 1. 名称：即操作的名称，必须简单可读性⾼，它应该是⼀个抽像通⽤的标识，不能太具体。 2. SpanId：当调⽤中唯⼀ID 3. ParentId：表示其⽗Span 4. 开始与结束时间

端到端Span

一次远程调用需要记录几个Span呢？

我们需要在客户端和服务端分别记录Span信息，这样才能计在两个端的视角分别记录信息。比如计算中间的网络IO。

在Dapper 中分布式请求起码包含如下四个核⼼埋点阶段：

1. 客户端发送 cs（Client Send）：客户端发起请求时埋点，记录客户端发起请求的时间戳

2. 服务端接收 sr（Server Receive）：服务端接受请求时埋点，记录服务端接收到请求的时间戳

3. 服务端响应 ss（Server Send）：服务端返回请求时埋点，记录服务端响应请求的时间戳

4. 客户端接收 cr（Client Receive）：客户端接受返回结果时埋点，记录客户端接收到响应时的时间戳

通过这四个埋点信息，我们可以得到如下信息：

客户端请求服务端的网络耗时：sr-cs

服务端处理请求的耗时：ss-sr

服务端发送响应给客户端的网络耗时：cr-ss

本次请求在这两个服务之间的总耗时：cr-cs

以上这些埋点在 Dapper 中有个专业的术语，叫做 Annotation。如果 Dapper 论⽂中的图示你还没有看太懂的话，那么可以再看看下⾯这张图，⽐较清楚的展示出整个过程。

参考

Dapper论文：
https://research.google/pubs/pub36356/

Dapper大规模分布式系统跟踪基础设施论文：
https://storage.googleapis.com/pub-tools-public-publication-data/pdf/36356.pdf

作者：京东物流 崔旭

我们都知道，数据结构和算法本身解决的是“快”和“省”的问题，即如何让代码运行得更快，如何让代码更省存储空间。所以，执行效率是算法一个非常重要的考量指标。那如何来衡量你编写的算法代码的执行效率呢？这里就要用到我们今天要讲的内容：时间、空间复杂度分析。

1 为什么需要复杂度分析？

你可能会有些疑惑，我把代码跑一遍，通过统计、监控，就能得到算法执行的时间和占用的内存大小。为什么还要做时间、空间复杂度分析呢？这种分析方法能比实实在在跑一遍得到的数据更准确吗？

首先可以肯定地说，这种评估算法执行效率的方法是正确的。很多数据结构和算法书籍还给这种方法起了一个名字，叫事后统计法。但是，这种统计方法有非常大的局限性。

1.1 测试结果非常依赖测试环境

测试环境中硬件的不同会对测试结果有很大的影响。比如，我们拿同样一段代码，分别用 Intel Core i9 处理器和 Intel Core i3 处理器来运行，i9 处理器要比 i3 处理器执行的速度快很多。还有，比如原本在这台机器上 a 代码执行的速度比 b 代码要快，等我们换到另一台机器上时，可能会有截然相反的结果。

1.2 测试结果受数据规模的影响很大

对同一个排序算法，待排序数据的有序度不一样，排序的执行时间就会有很大的差别。极端情况下，如果数据已经是有序的，那排序算法不需要做任何操作，执行时间就会非常短。除此之外，如果测试数据规模太小，测试结果可能无法真实地反应算法的性能。比如，对于小规模的数据排序，插入排序可能反倒会比快速排序要快！

所以，我们需要一个不用具体的测试数据来测试，就可以粗略地估计算法的执行效率的方法，这就是我们接下来要说的大O复杂度表示法。

2 大O复杂度表示法

算法的执行效率，粗略地讲，就是算法代码执行的时间。但是，如何在不运行代码的情况下，用“肉眼”得到一段代码的执行时间呢？

这里有段非常简单的代码，求 1,2,3…n 的累加和。现在，一块来估算一下这段代码的执行时间吧。

从 CPU 的角度来看，这段代码的每一行都执行着类似的操作：读数据-运算-写数据。尽管每行代码对应的 CPU 执行的个数、执行的时间都不一样，但是，我们这里只是粗略估计，所以可以假设每行代码执行的时间都一样，为 unit_time。在这个假设的基础之上，这段代码的总执行时间是多少呢？

第 2、3 行代码分别需要 1 个 unit_time 的执行时间，第 4、5 行都运行了 n 遍，所以需要 2n_unit_time 的执行时间，所以这段代码总的执行时间就是 (2n+2)_unit_time。可以看出来，所有代码的执行时间 T(n) 与每行代码的执行次数成正比。

按照这个分析思路，我们再来看这段代码。

我们依旧假设每个语句的执行时间是 unit_time。那这段代码的总执行时间 T(n) 是多少呢？

第 2、3、4 行代码，每行都需要 1 个 unit_time 的执行时间，第 5、6 行代码循环执行了 n 遍，需要 2n_unit_time 的执行时间，第 7、8 行代码循环执行了 n²遍，所以需要 2n²_unit_time 的执行时间。所以，整段代码总的执行时间 T(n) = (2n²+2n+3)*unit_time。

尽管我们不知道 unit_time 的具体值，但是通过这两段代码执行时间的推导过程，我们可以得到一个非常重要的规律，那就是，所有代码的执行时间 T(n) 与每行代码的执行次数 n 成正比。我们可以把这个规律总结成一个公式。注意，大 O 就要登场了！

我来具体解释一下这个公式。其中，T(n) 我们已经讲过了，它表示代码执行的时间；n 表示数据规模的大小；f(n) 表示每行代码执行的次数总和。因为这是一个公式，所以用 f(n) 来表示。公式中的 O，表示代码的执行时间 T(n) 与 f(n) 表达式成正比。

所以，第一个例子中的 T(n) = O(2n+2)，第二个例子中的 T(n) = (2n²+2n+3)。这就是大O时间复杂度表示法。大O时间复杂度实际上并不具体表示代码真正的执行时间，而是表示代码执行时间随数据规模增长的变化趋势，所以，也叫作渐进时间复杂度，简称时间复杂度。

当 n 很大时，你可以把它想象成 10000、100000。而公式中的低阶、常量、系数三部分并不左右增长趋势，所以都可以忽略。我们只需要记录一个最大量级就可以了，如果用大 O 表示法表示刚讲的那两段代码的时间复杂度，就可以记为：T(n) = O(n)； T(n) = O(n²)。

3 时间复杂度分析

前面介绍了大 O 时间复杂度的由来和表示方法。现在我们来看下，如何分析一段代码的时间复杂度？

3.1 只关注循环执行次数最多的一段代码

大 O 这种复杂度表示方法只是表示一种变化趋势。我们通常会忽略掉公式中的常量、低阶、系数，只需要记录一个最大阶的量级就可以了。所以，我们在分析一个算法、一段代码的时间复杂度的时候，也只关注循环执行次数最多的那一段代码就可以了。这段核心代码执行次数的 n 的量级，就是整段要分析代码的时间复杂度。

为了便于你理解，我还拿前面的例子来说明。

其中第 2、3 行代码都是常量级的执行时间，与 n 的大小无关，所以对于复杂度并没有影响。循环执行次数最多的是第 4、5 行代码，所以这块代码要重点分析。前面我们也讲过，这两行代码被执行了 n 次，所以总的时间复杂度就是 O(n)。

3.2 加法法则：总复杂度等于量级最大的那段代码的复杂度

这里还有一段代码。

这个代码分为三部分，分别是求 sum_1、sum_2、sum_3。我们可以分别分析每一部分的时间复杂度，然后把它们放到一块儿，再取一个量级最大的作为整段代码的复杂度。

第一段的时间复杂度是多少呢？这段代码循环执行了 100 次，所以是一个常量的执行时间，跟 n 的规模无关。

即便这段代码循环 10000 次、100000 次，只要是一个已知的数，跟 n 无关，照样也是常量级的执行时间。当 n 无限大的时候，就可以忽略。尽管对代码的执行时间会有很大影响，但是回到时间复杂度的概念来说，它表示的是一个算法执行效率与数据规模增长的变化趋势，所以不管常量的执行时间多大，我们都可以忽略掉。因为它本身对增长趋势并没有影响。

那第二段代码和第三段代码的时间复杂度是多少呢？答案是 O(n) 和 O(n²)。
综合这三段代码的时间复杂度，我们取其中最大的量级。所以，整段代码的时间复杂度就为 O(n²)。也就是说：总的时间复杂度就等于量级最大的那段代码的时间复杂度。那我们将这个规律抽象成公式就是：

如果 T1(n)=O(f(n))，T2(n)=O(g(n))；那么 T(n)=T1(n)+T2(n)=max(O(f(n)), O(g(n))) =O(max(f(n), g(n))).

3.3 乘法法则：嵌套代码的复杂度等于嵌套内外代码复杂度的乘积

刚讲了一个复杂度分析中的加法法则，这儿还有一个乘法法则。类比一下，你应该能“猜到”公式是什么样子的吧？

如果 T1(n)=O(f(n))，T2(n)=O(g(n))；那么 T(n)=T1(n)_T2(n)=O(f(n))_O(g(n))=O(f(n)*g(n)).

也就是说，假设 T1(n) = O(n)，T2(n) = O(n²)，则 T1(n) * T2(n) = O(n³)。落实到具体的代码上，我们可以把乘法法则看成是嵌套循环，我举个例子给你解释一下。

我们单独看 cal() 函数。假设 f() 只是一个普通的操作，那第 4～6 行的时间复杂度就是，T1(n) = O(n)。但 f() 函数本身不是一个简单的操作，它的时间复杂度是 T2(n) = O(n)，所以，整个 cal() 函数的时间复杂度就是，T(n) = T1(n)_T2(n) = O(n_n) = O(n²)。

3.4 几种常见时间复杂度实例分析

虽然代码千差万别，但是常见的复杂度量级并不多。稍微总结了一下，这些复杂度量级几乎涵盖了大部分的场景。

• 常量阶 O(1)
• 对数阶 O(logn)
• 线性阶 O(n)
• 线性对数阶 O(nlogn)
• 平方阶 O(n²)
• 立方阶 O(n³) …
• 指数阶 O(2ⁿ)
• 阶乘阶 O(n!)

对于刚罗列的复杂度量级，我们可以粗略地分为两类，多项式量级和非多项式量级。其中，非多项式量级只有两个：O(2ⁿ) 和 O(n!)。

当数据规模 n 越来越大时，非多项式量级算法的执行时间会急剧增加，求解问题的执行时间会无限增长。所以，非多项式时间复杂度的算法其实是非常低效的算法。我们主要来看几种常见的多项式时间复杂度。

1.O(1)

首先你必须明确一个概念，O(1) 只是常量级时间复杂度的一种表示方法，并不是指只执行了一行代码。比如这段代码，即便有 3 行，它的时间复杂度也是 O(1），而不是 O(3)。

只要代码的执行时间不随 n 的增大而增长，这样代码的时间复杂度我们都记作 O(1)。或者说，一般情况下，只要算法中不存在循环语句、递归语句，即使有成千上万行的代码，其时间复杂度也是Ο(1)。

2.O(logn)、O(nlogn)

对数阶时间复杂度非常常见，同时也是最难分析的一种时间复杂度。我通过一个例子来说明一下。

根据我们前面讲的复杂度分析方法，第三行代码是循环执行次数最多的。所以，我们只要能计算出这行代码被执行了多少次，就能知道整段代码的时间复杂度。
从代码中可以看出，变量 i 的值从 1 开始取，每循环一次就乘以 2。当大于 n 时，循环结束。

实际上，变量 i 的取值就是一个等比数列。如果我把它一个一个列出来，就应该是这个样子的：

所以，我们只要知道 x 值是多少，就知道这行代码执行的次数了。通过 2ˣ=n 求解 x ，x=log₂n，所以，这段代码的时间复杂度就是 O(log₂n)。

现在，我把代码稍微改下，你再看看，这段代码的时间复杂度是多少？

根据我刚刚讲的思路，很简单就能看出来，这段代码的时间复杂度为 O(log₃n)。

实际上，不管是以 2 为底、以 3 为底，还是以 10 为底，我们可以把所有对数阶的时间复杂度都记为 O(logn)。为什么呢？

我们知道，对数之间是可以互相转换的，log₃n 就等于 log₃2_log₂n，所以 O(log₃n) = O(C_log₂n)，其中 C=log₃2 是一个常量。基于我们前面的一个理论：在采用大 O 标记复杂度的时候，可以忽略系数，即 O(Cf(n)) = O(f(n))。所以，O(log₂n) 就等于 O(log₃n)。因此，在对数阶时间复杂度的表示方法里，我们忽略对数的“底”，统一表示为 O(logn)。

如果你理解了O(logn)，那 O(nlogn) 就很容易理解了。还记得我们刚讲的乘法法则吗？如果一段代码的时间复杂度是 O(logn)，我们循环执行 n 遍，时间复杂度就是 O(nlogn) 了。而且，O(nlogn) 也是一种非常常见的算法时间复杂度。比如，归并排序、快速排序的时间复杂度都是 O(nlogn)。

3.O(m+n)、O(m*n)

我们再来讲一种跟前面都不一样的时间复杂度，代码的复杂度由两个数据的规模来决定。老规矩，先看代码！

从代码中可以看出，m 和 n 是表示两个数据规模。我们无法事先评估 m 和 n 谁的量级大，所以我们在表示复杂度的时候，就不能简单地利用加法法则，省略掉其中一个。所以，上面代码的时间复杂度就是 O(m+n)。

针对这种情况，原来的加法法则就不正确了，我们需要将加法规则改为：T1(m) + T2(n) = O(f(m) + g(n))。但是乘法法则继续有效：T1(m)_T2(n) = O(f(m)_f(n))。

4 空间复杂度分析

前面，咱们花了很长时间讲大 O 表示法和时间复杂度分析，理解了前面讲的内容，空间复杂度分析方法学起来就非常简单了。

时间复杂度的全称是渐进时间复杂度，表示算法的执行时间与数据规模之间的增长关系。类比一下，空间复杂度全称就是渐进空间复杂度，表示算法的存储空间与数据规模之间的增长关系。

还是拿具体的例子来说明。

跟时间复杂度分析一样，我们可以看到，第 2 行代码中，我们申请了一个空间存储变量 i，但是它是常量阶的，跟数据规模 n 没有关系，所以我们可以忽略。第 3 行申请了一个大小为 n 的 int 类型数组，除此之外，剩下的代码都没有占用更多的空间，所以整段代码的空间复杂度就是 O(n)。
我们常见的空间复杂度就是 O(1)、O(n)、O(n²)，像 O(logn)、O(nlogn) 这样的对数阶复杂度平时都用不到。而且，空间复杂度分析比时间复杂度分析要简单很多。

5 内容小结

复杂度也叫渐进复杂度，包括时间复杂度和空间复杂度，用来分析算法执行效率与数据规模之间的增长关系，可以粗略地表示，越高阶复杂度的算法，执行效率越低。常见的复杂度并不多，从低阶到高阶有：O(1)、O(logn)、O(n)、O(nlogn)、O(n²)。

aliases: []
tags   : " #QA #Java "
summary: [POI生成Excel超出的单元格样式的最大数量]
author : [yaenli]
notekey: [20230322-100908]

问题现象

使用Apache POI生成Excel时，如果创建的单元格样式过多，会报样式超出最大数的错误，

.xls
的异常错误：

java.lang.IllegalStateException: The maximum number of cell styles was exceeded. You can define up to 4000 styles in a .xls workbook
	at org.apache.poi.hssf.usermodel.HSSFWorkbook.createCellStyle(HSSFWorkbook.java:1144)
	at org.apache.poi.hssf.usermodel.HSSFWorkbook.createCellStyle(HSSFWorkbook.java:88)

.xlsx
的异常错误：

java.lang.IllegalStateException: The maximum number of Cell Styles was exceeded. You can define up to 64000 style in a .xlsx Workbook
	at org.apache.poi.xssf.model.StylesTable.createCellStyle(StylesTable.java:830)
	at org.apache.poi.xssf.usermodel.XSSFWorkbook.createCellStyle(XSSFWorkbook.java:750)

问题分析

同一个Workbook创建CellStyle有最大数限制，其中
.xls（Excel 97）
的最大数是4000，
.xlsx（Excel 2007）
的最大数是64000 。

xls
参数限制于
org.apache.poi.hssf.usermodel.HSSFWorkbook
:

private static final int MAX_STYLES = 4030;

public HSSFCellStyle createCellStyle() {
    if (this.workbook.getNumExFormats() == MAX_STYLES) {
      throw new IllegalStateException("The maximum number of cell styles was exceeded. You can define up to 4000 styles in a .xls workbook");
    }
    
    ExtendedFormatRecord xfr = this.workbook.createCellXF();
    short index = (short)(getNumCellStyles() - 1);
    return new HSSFCellStyle(index, xfr, this);
}

xlsx
参数限制于
org.apache.poi.xssf.model.StylesTable
:

private static final int MAXIMUM_STYLE_ID = SpreadsheetVersion.EXCEL2007.getMaxCellStyles();// 64000

public XSSFCellStyle createCellStyle() {
    if (getNumCellStyles() > MAXIMUM_STYLE_ID) {
      throw new IllegalStateException("The maximum number of Cell Styles was exceeded. You can define up to " + MAXIMUM_STYLE_ID + " style in a .xlsx Workbook");
    }
    
    int xfSize = this.styleXfs.size();
    CTXf xf = CTXf.Factory.newInstance();
    xf.setNumFmtId(0L);
    xf.setFontId(0L);
    xf.setFillId(0L);
    xf.setBorderId(0L);
    xf.setXfId(0L);
    int indexXf = putCellXf(xf);
    return new XSSFCellStyle(indexXf - 1, xfSize - 1, this, this.theme);
}

因此，在生成Excel时，如果同一个Workbook不停的创建CellStyle，超限时就会产生样式最大数异常，最直接的体现就是在某些代码中，对每个单元格去独立的设置样式，生成大数据量的Excel报错。

解决方案

网上最热门的解决方案是所谓的将
createCellStyle
放在循环外面，这只能应付表格样式单一的情况。

由于单元格样式
CellStyle
并不是单元独立拥有的，每个单元格只是保存了样式的索引，一般的Excel真正使用到的样式也不会超过
4000/64000
，因此更好的解决方案是实现单元格样式的复用（
注意不同的Workbook创建的CellStyle是不能混用的
）。

方案1：缓存样式实现复用

提取样式关键字作为key，将CellStyle缓存至Map：

Workbook workBook = new HSSFWorkbook();// or new XSSFWorkbook();
Sheet sheet = workBook.createSheet(strSheetName);
Map cellStyleMap = new HashMap<String, CellStyle>();// 缓存样式

// 样式代码
for (int rowIndex = 0; rowIndex < maxRow; rowIndex++) {
    Row row = sheet.createRow(rowIndex);
    for (int colIndex = 0; colIndex < maxCol; colIndex++) {
        Cell cell = row.createCell((short) colIndex);
        String styKey = getCellStyleKey(rowIndex, colIndex);// 根据获取样式key
        CellStyle cellStyle = (CellStyle) cellStyleMap.computeIfAbsent(styKey, k-> workBook.createCellStyle());// 获取样式
        cell.setCellStyle(cellStyle);
    }
}

方案2：修改限制参数

修改POI中的限制参数（
org.apache.poi.hssf.usermodel.HSSFWorkbook.MAX_STYLES
或
org.apache.poi.ss.SpreadsheetVersion.EXCEL2007
）。

过多的创建样式会影响性能，建议仅在真正使用的样式超过限制时再去修改此参数。

方案3：延迟指定单元格样式实现复用

参见文章：
POI 操作Excel的单元格样式超过64000的异常问题解决
根据模版填充Excel并导出的工具 · GitCode

一、背景

Suricata支持网卡在线抓包和离线读取PCAP包两种形式的抓包：

• 离线抓包天然具有速度慢、非实时的特点
• 在线捕获数据包又包括常规网卡抓包、PF_RING和DPDK的方式

由于项目分光的流量较大, 软件自带的抓包方式并不能满足需求，因此采用了基于DPDK的Suricata在线捕获网卡数据包的方式。

二、 服务器配置与对应软件版本

操作系统：Centos7.6 1810内核版本：3.10.0-1160.88.1.el7.x86_64

网卡信息：Intel X722 万兆光口以太网卡

DPDK版本：dpdk-19.11.14Suricata版本：DPDK_Suricata-4.1.4

三、部署安装DPDK

基于DPDK抓包的Suricata版本只更新到4.1.4，因此对DPDK版本有要求，经过测试推荐使用
DPDK-19.11.14

DPDK官网下载地址：
http://fast.dpdk.org/rel/dpdk-19.11.14.tar.gz

1、安装dpdk-19.11.14需要操作系统内核版本大于3.2，如果版本过低，可以通过以下方式升级

1 [root@ids-dpdk ~]# cat /etc/redhat-release2 CentOS Linux release 7.6.1810(Core)3 [root@ids-dpdk ~]# uname -r #查看一下系统版本4 3.10.0-957.el7.x86_645 [root@ids-dpdk ~]# rpm -qa kernel #通过rpm命令查看我所安装的内核版本6 kernel-3.10.0-957.el7.x86_647 [root@ids-dpdk ~]# ls /usr/src/kernels/#查看有没有相应的内核开发包8空 #如果该目录下没有系统内核源码，执行以下操作9 [root@ids-dpdk ~]# yum install kernel-devel #安装内核头文件后10 [root@ids-dpdk ~]# ls /usr/src/kernels/
11 3.10.0-1160.88.1.el7.x86_6412#两个版本号不一致，进行升级并重启13 [root@ids-dpdk ~]# yum -y update kernel kernel-devel14[root@ids-dpdk ~]# reboot15#再此查看，版本号一致，问题解决16 [root@ids-dpdk ~]# ls /usr/src/kernels/
17 3.10.0-1160.88.1.el7.x86_6418 [root@ids-dpdk ~]# uname -r19 3.10.0-1160.88.1.el7.x86_64

2、安装依赖包

sudo yum install -y gcc make

sudo yum install-y libpcap libpcap-devel

sudo yum install-y numactl numactl-devel
sudo yum install -y pciutils

3、从dpdk官网下载dpdk压缩包并解压

将dpdk压缩包下载到/home目录下并解压

wget http://fast.dpdk.org/rel/dpdk-19.11.14.tar.gz
tar -zxvf dpdk-19.11.14.tar.gz

4、DPDK编译和网卡绑定

1. 设置环境变量，命令行执行：

[root@ids-dpdk ~]# export RTE_SDK='/home/dpdk-19.11.14'[root@ids-dpdk ~]# export RTE_TARGET=x86_64-native-linuxapp-gcc

#(对于64位机用这个命令，对于32位机用i686-native-linuxapp-gcc)2. 查看环境变量是否设置好：

[root@ids-dpdk ~]# env |grep RTE

RTE_SDK=/home/dpdk-stable-19.11.14RTE_TARGET=x86_64-native-linuxapp-gcc3. 关闭要绑定的网卡，否则绑定dpdk时不成功

ifconfig  ens1f0 down4. 进入到dpdk-19.11.14/usertools目录下

cd/home/dpdk-19.11.14/usertools

执行./dpdk-setup.sh



会输出一列可选操作：------------------------------------------------------------------------------RTE_SDK exportedas /home/dpdk-stable-19.11.14
------------------------------------------------------------------------------
----------------------------------------------------------Step1: Select the DPDK environment to build----------------------------------------------------------[1] arm64-armada-linuxapp-gcc

[2] arm64-armada-linux-gcc

[3] arm64-armv8a-linuxapp-clang

[4] arm64-armv8a-linuxapp-gcc

[5] arm64-armv8a-linux-clang

[6] arm64-armv8a-linux-gcc

[7] arm64-bluefield-linuxapp-gcc

[8] arm64-bluefield-linux-gcc

[9] arm64-dpaa-linuxapp-gcc

[10] arm64-dpaa-linux-gcc

[11] arm64-emag-linuxapp-gcc

[12] arm64-emag-linux-gcc

[13] arm64-graviton2-linuxapp-gcc

[14] arm64-graviton2-linux-gcc

[15] arm64-n1sdp-linuxapp-gcc

[16] arm64-n1sdp-linux-gcc

[17] arm64-octeontx2-linuxapp-gcc

[18] arm64-octeontx2-linux-gcc

[19] arm64-stingray-linuxapp-gcc

[20] arm64-stingray-linux-gcc

[21] arm64-thunderx2-linuxapp-gcc

[22] arm64-thunderx2-linux-gcc

[23] arm64-thunderx-linuxapp-gcc

[24] arm64-thunderx-linux-gcc

[25] arm64-xgene1-linuxapp-gcc

[26] arm64-xgene1-linux-gcc

[27] arm-armv7a-linuxapp-gcc

[28] arm-armv7a-linux-gcc

[29] graviton2

[30] i686-native-linuxapp-gcc

[31] i686-native-linuxapp-icc

[32] i686-native-linux-gcc

[33] i686-native-linux-icc

[34] ppc_64-power8-linuxapp-gcc

[35] ppc_64-power8-linux-gcc

[36] x86_64-native-bsdapp-clang

[37] x86_64-native-bsdapp-gcc

[38] x86_64-native-freebsd-clang

[39] x86_64-native-freebsd-gcc

[40] x86_64-native-linuxapp-clang

[41] x86_64-native-linuxapp-gcc

[42] x86_64-native-linuxapp-icc

[43] x86_64-native-linux-clang

[44] x86_64-native-linux-gcc

[45] x86_64-native-linux-icc

[46] x86_x32-native-linuxapp-gcc

[47] x86_x32-native-linux-gcc----------------------------------------------------------Step2: Setup linux environment----------------------------------------------------------[48] Insert IGB UIO module

[49] Insert VFIO module

[50] Insert KNI module

[51] Setup hugepage mappings for non-NUMA systems

[52] Setup hugepage mappings forNUMA systems

[53] Display current Ethernet/Baseband/Crypto device settings

[54] Bind Ethernet/Baseband/Crypto device to IGB UIO module

[55] Bind Ethernet/Baseband/Crypto device to VFIO module

[56] Setup VFIO permissions----------------------------------------------------------Step3: Run test application forlinux environment----------------------------------------------------------[57] Run test application ($RTE_TARGET/app/test)

[58] Run testpmd application in interactive mode ($RTE_TARGET/app/testpmd)----------------------------------------------------------Step4: Other tools----------------------------------------------------------[59] List hugepage info from /proc/meminfo----------------------------------------------------------Step5: Uninstall and system cleanup----------------------------------------------------------[60] Unbind devices fromIGB UIO or VFIO driver

[61] Remove IGB UIO module

[62] Remove VFIO module

[63] Remove KNI module

[64] Remove hugepage mappings



[65] Exit Script



Option: 



在最下面的 Option: 处输入41会选择适合x86_64机器的gcc编译器，如果是其他架构的机器，需要选择对应的其他编译器

编译中，稍等片刻......

编译完成后会出现

Build complete [x86_64-native-linuxapp-gcc]

Installation cannot run with T defined and DESTDIR undefined------------------------------------------------------------------------------RTE_TARGET exportedas x86_64-native-linuxapp-gcc------------------------------------------------------------------------------Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:48加载igb uio模块

Unloading any existing DPDK UIO module

Loading DPDK UIO module



Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:52配置大页

Removing currently reserved hugepages

Unmounting/mnt/huge and removing directory



  Input the number of 2048kB hugepagesforeach node

  Example: to have 128MB of hugepages available per nodeina 2MB huge page system,

  enter'64' to reserve 64 *2MB pages on each node

Number of pagesfor node0: 1024（在此输入1024）



Reserving hugepages

Creating/mnt/huge and mounting ashugetlbfs



Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:53列出所有网卡

Network devicesusingkernel driver===================================
0000:17:00.0 'MT27800 Family [ConnectX-5] 1017' if=ens5f0 drv=mlx5_core unused=igb_uio *Active*
0000:17:00.1 'MT27800 Family [ConnectX-5] 1017' if=ens5f1 drv=mlx5_core unused=igb_uio *Active*
0000:33:00.0 'Ethernet Connection X722 for 10GbE SFP+ 0dda' if=ens1f0 drv=i40e unused=igb_uio0000:33:00.1 'Ethernet Connection X722 for 10GbE SFP+ 0dda' if=ens1f1 drv=i40e unused=igb_uio



No'Baseband'devices detected==============================No'Crypto'devices detected============================No'Eventdev'devices detected==============================No'Mempool'devices detected=============================No'Compress'devices detected==============================No'Misc (rawdev)'devices detected===================================Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:54绑定网卡

Network devicesusingkernel driver===================================
0000:17:00.0 'MT27800 Family [ConnectX-5] 1017' if=ens5f0 drv=mlx5_core unused=igb_uio *Active*
0000:17:00.1 'MT27800 Family [ConnectX-5] 1017' if=ens5f1 drv=mlx5_core unused=igb_uio *Active*
0000:33:00.0 'Ethernet Connection X722 for 10GbE SFP+ 0dda' if=ens1f0 drv=i40e unused=igb_uio0000:33:00.1 'Ethernet Connection X722 for 10GbE SFP+ 0dda' if=ens1f1 drv=i40e unused=igb_uio



No'Baseband'devices detected==============================No'Crypto'devices detected============================No'Eventdev'devices detected==============================No'Mempool'devices detected=============================No'Compress'devices detected==============================No'Misc (rawdev)'devices detected===================================Enter PCI address of device to bind to IGB UIO driver:0000:33:00.0（在此输入网卡pci号，就是上边列出来的网卡信息）



成功后会输出：

ok



Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:53查看网卡绑定情况

Network devicesusing DPDK-compatible driver============================================
0000:33:00.0 'Ethernet Connection X722 for 10GbE SFP+ 0dda' drv=igb_uio unused=i40e



Network devicesusingkernel driver===================================
0000:17:00.0 'MT27800 Family [ConnectX-5] 1017' if=ens5f0 drv=mlx5_core unused=igb_uio  *Active*
0000:17:00.1 'MT27800 Family [ConnectX-5] 1017' if=ens5f1 drv=mlx5_core unused=igb_uio  *Active*
0000:33:00.1 'Ethernet Connection X722 for 10GbE SFP+ 0dda' if=enp51s0f1 drv=i40e unused=igb_uio 



No'Baseband'devices detected==============================No'Crypto'devices detected============================No'Eventdev'devices detected==============================No'Mempool'devices detected=============================No'Compress'devices detected==============================No'Misc (rawdev)'devices detected===================================Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:57进行简单测试：

  Enter hex bitmask of cores to execute test app on

  Example: to execute app on cores0 to 7, enter 0xffbitmask:0xff（在此输入 0xff）

Launching app

EAL: Detected40lcore(s)

EAL: Detected2NUMA nodes

EAL: Multi-process socket /var/run/dpdk/rte/mp_socket

EAL: Selected IOVA mode'PA'EAL: No available hugepages reportedin hugepages-1048576kB

EAL: Probing VFIO support...

EAL: PCI device0000:00:04.0 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.1 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.2 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.3 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.4 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.5 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.6 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.7 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:3d:00.0 on NUMA socket 0EAL:   probe driver:8086:37d3 net_i40e

i40e_GLQF_reg_init(): i40e device0000:3d:00.0 changed global register [0x002689a0]. original: 0x0000002a, new: 0x00000029i40e_GLQF_reg_init(): i40e device0000:3d:00.0 changed global register [0x00268ca4]. original: 0x00002826, new: 0x00009420EAL: PCI device0000:3d:00.1 on NUMA socket 0EAL:   probe driver:8086:37d3 net_i40e

EAL: PCI device0000:80:04.0 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.1 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.2 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.3 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.4 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.5 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.6 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.7 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

APP: HPETis not enabled, using TSC as defaulttimer

输出以上内容表示安装dpdk成功

RTE>>quit（在此输入quit）



Press enter tocontinue....



按下回车键，继续选择其他操作：

Option:58进行抓包测试

  Enter hex bitmask of cores to execute testpmd app on

  Example: to execute app on cores0 to 7, enter 0xffbitmask:7（输入 7）

Launching app

EAL: Detected40lcore(s)

EAL: Detected2NUMA nodes

EAL: Multi-process socket /var/run/dpdk/rte/mp_socket

EAL: Selected IOVA mode'PA'EAL: No available hugepages reportedin hugepages-1048576kB

EAL: Probing VFIO support...

EAL: PCI device0000:00:04.0 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.1 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.2 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.3 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.4 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.5 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.6 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:00:04.7 on NUMA socket 0EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:3d:00.0 on NUMA socket 0EAL:   probe driver:8086:37d3 net_i40e

EAL: PCI device0000:3d:00.1 on NUMA socket 0EAL:   probe driver:8086:37d3 net_i40e

EAL: PCI device0000:80:04.0 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.1 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.2 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.3 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.4 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.5 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.6 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

EAL: PCI device0000:80:04.7 on NUMA socket 1EAL:   probe driver:8086:2021rawdev_ioat

Interactive-mode selected

testpmd: create anew mbuf pool <mbuf_pool_socket_0>: n=163456, size=2176, socket=0testpmd: preferred mempool ops selected: ring_mp_mc



Warning! port-topology=paired and odd forward ports number, the last port will pair with itself.



Configuring Port0 (socket 0)

Port0: F0:10:90:7E:FF:03Checking link statuses...

Done

testpmd>start （输入start开始抓包）

io packet forwarding- ports=1 - cores=1 - streams=1 -NUMA support enabled, MP allocation mode: native

Logical Core1 (socket 0) forwards packets on 1streams:

  RX P=0/Q=0 (socket 0) -> TX P=0/Q=0 (socket 0) peer=02:00:00:00:00:00io packet forwarding packets/burst=32nb forwarding cores=1 - nb forwarding ports=1port0: RX queue number: 1 Tx queue number: 1Rx offloads=0x0 Tx offloads=0x10000RX queue:0RX desc=256 - RX free threshold=32RX threshold registers: pthresh=0 hthresh=0  wthresh=0RX Offloads=0x0TX queue:0TX desc=256 - TX free threshold=32TX threshold registers: pthresh=32 hthresh=0  wthresh=0TX offloads=0x10000 - TX RS bit threshold=32testpmd>stop （输入stop停止抓包）

Telling cores to stop...

Waitingforlcores to finish...---------------------- Forward statistics for port 0  ----------------------RX-packets: 2685              RX-dropped: 0             RX-total: 2685TX-packets: 2686              TX-dropped: 0             TX-total: 2686
  ----------------------------------------------------------------------------

  +++++++++++++++ Accumulated forward statistics for all ports+++++++++++++++RX-packets: 2685              RX-dropped: 0             RX-total: 2685TX-packets: 2686              TX-dropped: 0             TX-total: 2686
  ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++Done.

testpmd>quit（在此输入quit）



Stopping port0...

Stopping ports...

Done



Shutting down port0...

Closing ports...

Done



Bye...



Press enter tocontinue...



按下回车键，继续选择其他操作：

Option:65退出安装程序，至此，dpdk安装成功。

5、DPDK_Suricata编译与安装

1、安装依赖

[root@ids-dpdk ~]# yum -y install gcc libpcap-devel pcre-devel libyaml-devel file-devel \

  zlib-devel jansson-devel nss-devel libcap-ng-devel libnet-devel tar make openssl openssl-devel \

  libnetfilter_queue-devel lua-devel PyYAML libmaxminddb-devel rustc cargo librdkafka-devel \

  lz4-devel libxml2 autoconf2、下载DPDK_Suricata安装包

##目前DPDK_Suricata项目作者已经到4.1.4，只是目录仍沿用4.1.1，直接克隆项目代码即可

[root@ids-dpdk ~]# git clone https://github.com/vipinpv85/DPDK_SURICATA-4_1_1

3、编译并安装DPDK_Suricata

进入DPDK_Suricata目录下：

[root@ids-dpdk ~]# cd /home/DPDK_SURICATA-4_1_1/suricata-4.1.4构建支持DPDK的配置脚本：

[root@ids-dpdk suricata-4.1.4]# autoconf

使用dpdk进行配置：

[root@ids-dpdk suricata-4.1.4]# ./configure --prefix=/usr --sysconfdir=/etc --localstatedir=/var --enable-dpdk  

编译DPDK_Suricata：

[root@ids-dpdk suricata-4.1.4]# make

安装DPDK_Suricata：

[root@ids-dpdk suricata-4.1.4]# make install

更新规则文件：

[root@ids-dpdk suricata-4.1.4]# make install-full

安装相关配置文件：

[root@ids-dpdk suricata-4.1.4]# make install-conf

6、测试运行抓包

修改配置文件suricata.yaml

[root@ids-dpdk ~]# vim /etc/suricata/suricata.yaml

搜索 DPDK关键词，修改input-output-map和interface两处参数值

将input-output-map修改成：input-output-map: ["0"]

将interface修改成：- interface: 0修改完成后执行 :wq 保存并退出

测试运行：

[root@ids-dpdk ~]# suricata

如果报以下错误，是库文件未找到的原因/usr/bin/suricata: error while loading shared libraries: libhtp.so.2: cannot open shared objectfile: No such file or directory

执行命令手动建立到libhtp.so.2的软链接

[root@ids-dpdk ~]# ln -s /usr/lib/libhtp.so.2 /lib64/libhtp.so.2然后执行命令开始运行dpdk抓包测试

[root@ids-dpdk ~]# suricata --dpdk22/3/2023 -- 15:36:51 - <Notice> -  --26. (protocol-ff)22/3/2023 -- 15:36:51 - <Notice> -  --27. (protocol-ff)22/3/2023 -- 15:36:51 - <Notice> -  --28. (protocol-ff)22/3/2023 -- 15:36:51 - <Notice> -  --29. (protocol-ff)22/3/2023 -- 15:36:51 - <Notice> -  --30. (protocol-ff)22/3/2023 -- 15:36:51 - <Notice> -  --31. (protocol-ff)22/3/2023 -- 15:36:51 - <Notice> -  addr_dst_match4_cnt 1 addr_src_match4_cnt 1 addr_dst_match6_cnt i1 addr_src_match6_cnt 1
22/3/2023 -- 15:36:51 - <Notice> -IPV422/3/2023 -- 15:36:51 - <Notice> -  0:ffffffff22/3/2023 -- 15:36:51 - <Notice> -  0:ffffffff22/3/2023 -- 15:36:51 - <Notice> - -----------------------
22/3/2023 -- 15:36:51 - <Notice> -IPV622/3/2023 -- 15:36:51 - <Notice> -  0-0-0-0:ffffffff-ffffffff-ffffffff-ffffffff22/3/2023 -- 15:36:51 - <Notice> -  0-0-0-0:ffffffff-ffffffff-ffffffff-ffffffff22/3/2023 -- 15:36:51 - <Notice> - -----------------------
22/3/2023 -- 15:36:51 - <Notice> -Source Port22/3/2023 -- 15:36:51 - <Notice> -  port:port2 (0:ffff)22/3/2023 -- 15:36:51 - <Notice> -Destiantion Port22/3/2023 -- 15:36:51 - <Notice> -  port:port2 (0:ffff)22/3/2023 -- 15:36:51 - <Notice> -  prio 3 
22/3/2023 -- 15:36:51 - <Notice> -  Port 0 RX-q (4) hence trying RSS22/3/2023 -- 15:36:51 - <Notice> -  rss_hf 3ef8, rss_key_len 0

22/3/2023 -- 15:36:52 - <Notice> - all 4 packet processing threads, 4management threads initialized, engine started.

至此，DPDK_Suricata开启DPDK抓包

7、DPDK脚本补充

由于服务器开关机会导致DPDK绑定的网卡会被默认解绑，为简化重新机械的绑定工作，通过shell脚本实现自动化DPDK绑定网卡。

编辑/etc/profile，设置环境变量

在done和unset i之间添加环境变量，执行:wq保存并退出，然后执行source使之生效

[root@ids-dpdk ~]# vim /etc/profile
添加内容：

[root@ids-dpdk ~]# source /etc/profile

创建DPDK绑定脚本dpdk-bind.sh，并赋予执行权限

[root@ids-dpdk usertools]# vim dpdk-bind.sh

[root@ids-dpdk usertools]# chmod -R 775 dpdk-bind.sh

dpdk-bind.sh脚本内容如下，根据自己需要修改网卡名称和路径信息

#!/bin/sh



#网卡名

uio=ens1f0



#需要绑定的驱动类型igb_uio或者vfio-pci 

pci_type=igb_uio



#挂载驱动

modprobe uio

insmod/home/dpdk-stable-19.11.14/x86_64-native-linuxapp-gcc/kmod/igb_uio.ko



#关闭网卡

ifconfig $uio down



#绑定网卡到igb_uio

python/home/dpdk-stable-19.11.14/usertools/dpdk-devbind.py --bind=$pci_type $uio

设置大页存储，将DPDK绑定脚本加入开机启动项中，编辑/etc/rc.local，根据自己需求修改路径等信息

#!/bin/bash

# THIS FILE IS ADDED FOR COMPATIBILITY PURPOSES

#

# Itishighly advisable to create own systemd services or udev rules

# to run scripts during boot instead ofusing thisfile.

#

# In contrast to previous versions due to parallel execution during boot

#thisscript will NOT be run after all other services.

#

# Please note that you must run'chmod +x /etc/rc.d/rc.local'to ensure

# thatthisscript will be executed during boot.



touch/var/lock/subsys/local



#设置大页存储

echo1024 > /sys/kernel/mm/hugepages/hugepages-2048kB/nr_hugepages

mkdir/mnt/huge > /dev/null 2>&1mount-t hugetlbfs nodev /mnt/huge



#网卡绑定DPDK

sh/home/dpdk-stable-19.11.14/usertools/dpdk-bind.sh > /dev/null 2>&1

编辑完成后，执行:wq保存并退出

Android中阻塞队列的应用有哪些

阻塞队列在 Android 中有很多应用，比如：

1. 线程池：线程池任务的执行就是基于一个阻塞队列，如果线程池任务已满，则任务需要等待阻塞队列中的其他任务完成。
2. Handler 消息队列：Handler 的消息队列也是一种阻塞队列。handler发送消息时，首先将消息加入到消息队列中，在空闲状态下 MessageQueue 队列是阻塞的状态，直到队列不为空，Looper 开始轮询 MessageQueue 里面的 Message。

阻塞队列支持两个核心方法，分别为：

• put(E e):将元素插入队尾，支持阻塞式插入，在容量无限制的情况下一直等待直到队列有空闲位置。
• take(): 获取且移除此队列的头部，在队列为空时，阻塞等待这个头部可用，并返回被取出的元素。

在 Java 中，阻塞队列的实现类有 ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue、PriorityBlockingQueue 等等。这些队列通过使用 Lock 和 Condition 来保证队列的线程安全。

阻塞队列的原理

是当阻塞队列中没有数据的时候，读线程将被阻塞；当阻塞队列已满的时候，写线程将被阻塞。阻塞队列通过内部锁来控制读和写操作的相互访问以及有序性，保证在多线程并发的时候，执行插入或读取操作是广泛可接受的，而不会引起任何问题。

线程池是维护一个工作线程的线程容器，用于优化线程的创建和销毁带来的开销。下面简单介绍一下线程池的底层实现原理。

线程池执行流程

1. 在初始化时，线程池会通过
   ThreadPoolExecutor
   构造函数赋值并创建核心线程池；
2. execute()
   方法将任务存放到阻塞队列中，如果有空闲线程，则取出最先进入队列的任务开始执行，否则加入等待中的队列中；
3. 当队列长度达到阈值时，不再继续增加，此时若仍有新任务提交，则以上空闲线程处理任务；若添加的线程数仍超过最大线程数时，则拒绝该任务。

参数说明：

以下是线程池最基本的四类参数。

• 核心线程数（corePoolSize）：线程池中的基本线程数。当任务提交后，分类讨论：

  
  
  • 若线程池中的线程数小于 corePoolSize，那么即使线程池中还有空余的线程，也会直接添加一个新线程去处理当前的任务；
  • 若线程池已经有了 corePoolSize 个线程，那么任务就被加入阻塞队列进行缓冲，等待有空闲的线程来处理；
  • 一次性只能创建 corePoolSize 个线程。

• 最大线程数（maximumPoolSize）：线程池中允许存在的最大线程数。当缓冲队列满时，对于大于 corePoolSize 的任务将会此项启动更多的 Thread 来处理用户请求，针对有界和无界不同表现。如有等待队列功能满了后会执行
  RejectedExecutionException
  。在我看来这个配置很鸡肋，往往在它小于
  corepoolsize
  时候有表现，大于情况视大于多少对线程池复用机制上友好而已。有不对的欢迎评论区讨论下；

• 队列容量（workQueue）：等待任务的队列容量。在调用 execute() 方法时，阻塞队列可以用以下三种类型之一：

  
  
  • FIFO（先进先出）队列，使用java.util.concurrent.LinkedBlockingQueue，默认无界限大小。因此 newFixedThreadPool() 和 newSingleThreadExecutor() 的线程数最多也可能会达到 Integer.MAX_VALUE，会导致OOM等问题。但你可指定其大小。
  • LIFO（后进先出）队列，使用java.util.concurrent.LinkedBlockingDeque，特点是新任务插到队尾上，但在 JDK7 中不再使用这个队列，因为它是不符合 Java 应用程序的行为模式的；
  • 优先级队列，使用java.util.concurrent.PriorityBlockingQueue 类实现，按照元素等级来排序，低等级对象会先被获取，对于高等级的任务可以提前执行。

• 空闲线程销毁时间（keepAliveTime）：线程池中没有任务执行时，即空闲状态下的线程没事可做，这些空闲的线程会自动的销毁，首先由coreThreadHandle（正常获得锁的线程）完成剩下工作,然后 wait coreHandle 释放锁资源，休眠到内部队列不为空或超时返回，如果休眠超时则表示该线程超时了，在低于 corePoolSize 的情况下 除非设置allowCoreThreadTimout=true，否则永远不会发生；

实例代码

public class TestThreadPoll implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "正在执行...");
    }
}

  public static void main(String[] args) {
      // 创建一个基本的线程池
      ExecutorService executor = Executors.newCachedThreadPool();

      for (int i = 0; i < 10; i++) {
          Runnable task = new TestThreadPoll();
          executor.execute(task);
      }
      executor.shutdown();
  }

上面的创建了一个基本的线程池，并进行了线程数的添加。

线程池排队机制

当线程池中所有的线程都在执行任务时，如果新提交的任务需要被执行，那么这个新的任务会进入到一个阻塞队列中排队等待。而ThreadPoolExecutor中有四种常见的阻塞队列：SynchronousQueue、LinkedBlockingQueue、ArrayBlockingQueue、PriorityBlockingQueue。

其中SynchronousQueue是没有任何储存功能的阻塞队列，它只负责一个任务的传递，即前一个任务将后一个任务提交给线程池后直接结束，而不是进入等待队列。由于该队列没有储存空间，所以当一个线程提交任务时，该队列会寻找与之对应的另一个线程来接收这个任务。如果此时线程池处于饱和状态，这个任务就有可能被丢弃。

LinkedBlockingQueue则是一个带储存功能的无界队列，也就是说，这个队列可以一直往里添加新的元素。因此，它的长度限制仅仅是由内存大小来控制，如果你的阻塞队列需要存储非常多的任务，那么最好选用这个类型的阻塞队列。

ArrayBlockingQueue也是一个带储存功能的有界队列，它的长度是预设好的，如果你试图往这个队列中添加超过预设长度的任务，那么新增的任务就必须要等待其他任务完成或者被移除后才能加入队列了，换而言之，在这种队列中，无法满足新任务的请求时会出现阻塞情况。

PriorityBlockingQueue则是一个支持优先级排序的无界队列，也就是说，任务可以通过实现Comparable接口来设置一个队列中的优先级关系。当使用优先级队列时，可以为不同的任务设置不同的优先级，程序会按照任务的优先级顺序执行具体的任务。

自实现一个线程池实例

以下是一个简单的自实现线程池示例：

public class MyThreadPool {
    private final BlockingQueue<Runnable> workQueue;
    private final WorkerThread[] workers;

    public MyThreadPool(int nThreads) {
        this.workQueue = new LinkedBlockingQueue<>();
        this.workers = new WorkerThread[nThreads];

        for (int i = 0; i < nThreads; i++) {
            workers[i] = new WorkerThread();
            workers[i].start();
        }
    }

    public void execute(Runnable task) {
        synchronized (workQueue) {
            workQueue.add(task);
            workQueue.notify();
        }
    }

    private class WorkerThread extends Thread {
        @Override
        public void run() {
            while(true) {
                Runnable task;
                synchronized (workQueue) {
                    while (workQueue.isEmpty()) {
                        try {
                            workQueue.wait();
                        } catch (InterruptedException e) {
                            return;
                        }
                    }
                    task = workQueue.poll();
                }
                try {
                    task.run();
                } catch(Throwable t) {
                    // error handling code
                }
            }
        }
    }
}

这是一个最简单的线程池实现，其核心思想就是：在.WorkerThread类中定义了一个死循环，负责从任务队列中取出任务并执行。如果目前没有任务，则等待新任务到来。

使用时初始化线程池：

MyThreadPool pool = new MyThreadPool(2);

之后将需要做的任务加入线程池中：

pool.execute(new Runnable() {
    @Override
    public void run() {
        // 执行具体的任务代码
    }
});

至此，自实现的线程池实例就完成了。

execute框架解读

execute()
是Java中Executor接口的一个方法，它在执行传递给它的Runnable任务时使用。Executor框架提供了一种将任务提交给线程池以异步执行的方式。
当我们想要某段代码在异步环境中运行，又不需要知道每个任务何时完成时（所需时间可能会非常不同）的情况下，就可以使用这个框架。该框架管理和复用线程，以避免耗费创建他们的代价，并且在执行完任务后返回线程到线程池。
execute()
方法的目的是在默认池中安排一个任务来执行，由于这个方法只是将任务提交给线程池并立即返回，因此不能保证任务已经执行。这个方法只有在向线程池中添加任务时需要使用，例如对于没有返回值的简单操作或前置条件检查。
以下是
execute()
的方法语法：

void execute(Runnable command);

参数:

• command
  ：Runnable对象，该接口定义了需要在线程上执行的任务。

实现：
该方法直接将任务提交给主执行器以异步执行。然后返回而不等待任务的结束。如果需要使用结果，则调用
submit
方法。

//newFixedThreadPool 被重载用来覆盖默认设置，指定核心线程数和最大线程数相同。
//此示例将线程池大小限制为 5 个工作线程。
ExecutorService executor = Executors.newFixedThreadPool(5);

for (int i = 0; i < 10; i++) {
    Runnable worker = new ExampleRunnable("" + i);
    executor.execute(worker);
}
// 当所有可执行任务都完成后关闭线程池
executor.shutdown();
// 阻塞当前线程直到关闭操作完成
while (!executor.isTerminated()) { }

System.out.println("Finished all threads");

以上代码中，新的线程池有默认的限制，而不需要明确地声明线程池的大小或能力 。

线程池固定大小，当队列中没有可以在工作线程上执行的任务时（即阻止列表为空），则该任务将继续等待直到可用。

在每个可用工作线程上使用while循环反复启动一个Runnable任务并让其运行到池的关闭标志被设置

线程池被关闭之后，isTerminated()方法返回true，并退出while循环。