字符串多模式精确匹配（脏字/敏感词汇/关键字过滤算法）——TTMP算法 之实战F模式

前面那么多篇文章都太抽象，这次来一个稍微实际一点的。F模式是我实际上选用的模式，对该模式我做了不少实际的测试，因此代码也算是比较稳定的。不过由于
实际上为了得到该算法的效率，算法本身做了一些优化，对于初学者，理解起来可能会有点困难，因此不适合直接贴原始代码。为了便于大家理解，这里出示的代码
会比较好读，但是并不能取得我宣称的效果。大家可以在此基础上进行一定的优化，以便达到你所想要的性能。我目前版本的性能大约是：

T2080 1.73GHz（双核） 笔记本 1.5GB 内存 Vista Ultra 特性全开（集成显卡）

15M字符/秒

不过即便经过精简，这个算法的代码也还是比一般的算法要复杂：除了算法本身所在的类，它还需要5个额外的数据结构，以及一个特殊枚举的支持。这几个额外的数据结构包括：

FastDict

FastList

FastQueue

FastWorkItem

FastScanResult

一个额外的枚举是

CharacterType

其中前面四个数据类，以及那个枚举对于本算法是至关重要的。下面首先讲解一下这几个数据结构和枚举：



//
hashcode对应字符串列表的字典



public

class
FastDict : Dictionary
<
uint
, FastList
>







{



//
该hashcode对应的字符串最大长度是多少



public

int
MaxCharacterLength;



}





//
相同hashcode都有哪些字符串



public

class
FastList : List
<
string
>







{



}





//
一个起始字符记录（也就是可能需要处理的数据）



public

class
FastWorkItem







{



public
FastWorkItem(
int
index,
int
maxEndIndex,
uint
lowPartDelta)







{



Index
=
index;



MaxEndIndex
=
maxEndIndex;



LowPartDelta
=
lowPartDelta;



}



//
起始字符在待检文本中的位置



public

int
Index;



//
该起始字符所对应的关键字条目的最远结束位置



public

int
MaxEndIndex;



//
该记录所对应的（与前一条记录的）hashcode差值



public

uint
LowPartDelta;



}





//
待处理数据队列



public

class
FastQueue:Queue
<
FastWorkItem
>







{



}





//
扫描结果



public

class
FastScanResult







{



public
FastScanResult(
string
word,
int
index)







{



Word
=
word;



Index
=
index;



}



//
被扫描到的字符串



public

readonly

string
Word;



//



public

readonly

int
Index;



}





//
字符类型枚举



public

enum
CharacterType :
uint







{



//
该字符不是起始符，也不是中间的字符。（有可能是结束符，也可能不在脏字表当中）



NoHit
=

0
,



//
起始字符掩码，代表以该起始符起始的脏字条目的最大长度。



//
如果为0，表示不是起始符。



StarterMask
=

0x7fffffff
,



//
表示该字符是中间字符。



MiddleChar
=

0x80000000
,



}



其实，FastDict、FastList以及FastQueue本质上和.NET框架提供的没有什么差别——根本就没有什么代码嘛！没错，我主要是为了
书写简便，把这些个泛型给封闭了。虽然是没有什么代码，这几个结构却是算法要处理的核心数据。当我们初始化这个关键字过滤引擎的时候，就需要把所有的关键
字导入到FastDict/FastList数据结构，以及一个CharacterType数组当中。而当我们进行过滤的时候，就需要把一些待处理项转化
为FastWorkItem，并放入一个FastQueue队列中。那么我们是如何初始化这个引擎的呢？



public
MultiWordScanner(
string
[] scanKeys)







{



scanKeys
=
RemoveDuplicated(scanKeys);





//
Setup fast scan environment



//
Create a terminator indexer



FastDict[] result
=

new
FastDict[
char
.MaxValue];



//
Create a starter/ middle character indexer.



CharMap
=

new
CharacterType[
char
.MaxValue];



foreach
(
string
key
in
scanKeys)







{



//
create a dictionary for the specific terminator of this key



FastDict dict
=
CreateDict(result, key);



//
add the key into a list in this dictionary



CreateList(dict, key);



//
markup the middle character



MarkMiddleChar(key);



//
set the starter info



SetCharMap(key);



}



_fastIndex
=
result;



}





private
FastDict[] _fastIndex;

这段代码看着应该还算是简单，只是其中又是Dictionary又是List的，比较难搞明白其中的关系，之前的一些回复里面也有这样的问题。其实
啊，之前的“理论如此”篇章已经解说过一遍了，可能太抽象，不太好理解。这里呢，我就从Hashtable的理论知识开始说起，以便对Hashtable
还不是很了解的读者能够搞明白其中的关系。

当我们需要把一堆数据组织起来，比如说认为是一个集合，这个时候大家可以有很多的选择。比如说数组、链表、队列、栈、排序数等等，当然也包括了哈希表。我
们把这些数据组织起来，必然面临如何往里面添加一个元素，如何遍历每一个元素，如何找到某一个元素，如何删除一个元素等等问题。这些问题都涉及了两个效
率：空间效率和时间效率。比如说数组无疑是遍历速度最快的，存储空间最为节省的形式（不考虑压缩的情况下）。但是通常来讲，我们面临的问题并不是遍历所有
数据的速度，而是——例如在本任务当中，查找某一个元素的速度如何。那么哈希表是如何提高这方面的效率的呢？简单的讲，就是首先计算数据的特征值，然后根
据特征值来确定该数据应该存在于一个内部数组的什么位置（下标）。由于通常两个不同的数据，计算出来的特征值是不一样的，因此根据特征值来找到的位置通常
也就是我们想要的数据所在的位置。例如我们有A和BC两个数据，特征值的计算函数就是该数据的ASCII拼起来，那么有：

数据A，特征值：0x41000000

数据BC，特征值：0x42430000

而内部数组大小为11个元素，则

数据A可能所在的位置是：0x41000000 % 11 = 6

数据B可能所在的位置是：0x42430000 % 11 = 9

很显然，A和BC两个数据是可以放在位置6和位置9，而不会互相干扰的。但是，有的时候，不同的数据是有可能产生相同的特征值的。很容易想象嘛，如果说用
一个优先大小的特征值，就能够一一代表无穷尽的各种字符串组合，那么RAR就没有市场了，直接用特征值来“压缩”数据好了。同时，即便是不同的特征值，经
过一个模运算计算该特征值在内部数组的位置时，仍然可能会产生相同的位置结果。例如数据AB，特征值是0x41420000，算出来的位置也是6。那么这
个时候就叫做出现了冲突，有冲突就要解决，解决的办法有很多种。比如说再往下找一个空余的位置，或者干脆在该位置用一个列表来保存所有具有相同的位置计算
结果的数据。说道这里，大家都应该明白了，Hashtable在添加（查找）一个数据的时候，其步骤为：

1、计算数据特征值（也就是哈希值）

2、根据特征值计算下标

3、看看该位置是否为空余状态

4、对于添加，如果空余，就可以直接添加了，否则需要解决冲突；

对于查找，如果空余就返回没有结果，否则还要对比一下是否就是要找的数据：如果不是，则看一下是否有“冲突”的数据在后面，直到查找到没有冲突数据为止。

其实上面说的是Hashtable的一种特殊形式，叫做HashSet（.NET
3.5里面有提供）。比较正式的Hashtable，实际上添加和查询的时候，是通过一个key来操作的，上述的那些计算也不是对数据，而是对这个key
进行的。比如找到key为A的数据的过程，就变成通过计算key
A的特征值，再根据特征值找到可能的位置，然后通过比对所有的冲突直到找到"A"这个key，然后再把这个位置上的具体数据返回出来。

从上面这一大队的描述，我们基本上应该可以看出来，一个包含了n个数据的哈希表的查找时间速度接近于O(c)。也就是说，基本上是和数据的数量多少没有太
大关系的。而c的大小基本上是受下列因素影响的：哈希表本身的代码是否有效率；哈希值计算公式是否能有效的把数据特征值分散到每一个可能的数值上，可以想
象，不好的公式会导致大量数据有相同的哈希值，也就是会产生大量的冲突；解决冲突的机制是否有效率，有的解决方法可能会导致接下来添加的数据也会产生冲
突，或者找了半天还是没有找到空余的位置。

说了这半天好像还是没有解决你的困惑，别着急，马上就要说明白了：我们可以知道，由于存在冲突的可能性，在做查询的时候不能直接比对完特征值就确定该数据
一定存在，因此必不可少的一步就是做“数据比对”。也就是要看看，根据数据AB的特征值0x41420000计算出来的位置上面，是否真的存着"AB"这
个字符串。所以，如果我们直接使用HashTable或者HashSet，都会遇到一个问题：我们要直接提供一个数据"AB"，交由
HashTable/HashSet内部的算法来计算哈希值，以及进行冲突检查。因此，在我们的关键字过滤算法中，一旦触发“检索”，其代码就类似这样书
写：



for
(
int
length
=

1
; length
<=
maxLength; length
++
)







{



string
word
=
text.SubString(currentIndex, length);



if
(_hashTable.ContainsKey(word))







{



return

new
FastResult(word, currentIndex);



}



}



可以看到，这样的算法有这么几个问题（为了便于描述，我们假设在最糟糕的情况下）：

1、我们要分别计算从currentIndex开始，长度为1到maxLength的每一个字符串的哈希值；

2、我们要分别对从currentIndex开始，长度为1到maxLength的每一个情况，取出一个字符串对象，也就是不得不产生maxLength个字符串对象，也就不得不复制1+2+...+maxLength个字符。

实际上，这些基本上都是不必要的，或者说是可以被优化掉的。要不要优化，我们就要看一下这样的优化是否值得：对于我那个280K字符的正常文本，检出关键
字数量为1000个左右，但是触发检索的次数大概在10K到20K次左右。也就是说，如果采用上述算法，我们很可能至少得要为每一次检出关键字，多负担
10次的额外运算。为了减少这样可观数量的不必要运算，我简单的对哈希表进行了一个拆解——把计算哈希值，以及进行字符串对比（冲突检查）的工作给提取出
来，交由我的代码来进行运算。于是，就有了FastDict和FastList这两个基础数据结构了。

FastDict就是根据一个哈希值，给出该哈希值所对应的一个FastList。而FastList里面就保存着同一个哈希值所对应的所有字符串。换而
言之，FastDict就是根据数据的特征值，找到的可能的数据所在位置，FastList则是用来解决冲突的。通过这样的改造，我们就有机会对偏移量从
1到maxLength的字符
逐一扫描一遍，就可以分别得到长度为1到maxLength的maxLength个字符串的哈希值。同时，我们也不需要为每一次比较产生一个新的字符串
了。下面是其中一部分的代码：



public
FastScanResult FastScan(
string
text,
int
start)







{



FastQueue queue
=

new
FastQueue();



FastDict[] fastIndex
=
FastIndex;



//
上一次遇到起始符之前，计算出来的哈希值低16位部分。



uint
lastLowPart
=

0
;



//
当前处理项，该项永远不在queue中。



//
之所以这么处理，是为了避免大量的创建对象，和加入、弹出队列的操作



FastWorkItem currentItem
=

new
FastWorkItem(
-
1
,
-
1
,
0
);





int
count
=
text.Length;



for
(
int
i
=
start; i
<
count; i
++
)







{



uint
index
=
FastScanStringHelper.FastLower(text[i]);



CharacterType ctype
=
CharMap[index];



//
如果这个是起始符，那么下面这条语句则可以得出，



//
以该字符开头的关键字，除去起始符的最大长度是多少。



int
maxLength
=
(
int
) (ctype
&
CharacterType.StarterMask);



int
endIndex;



//
如果maxLength大于零，表示这是一个开始位置。



if
(maxLength
>

0
)







{



//
设置结束访问的位置，这和后面的To do 1，以及后面的代码都有关系



endIndex
=
i
+
maxLength
-

1
;



//
To do 1: 往currentItem或者queue中添加一条待解决起始符FastWorkItem记录



}



else







{



endIndex
=

0
;



}





//
如果当前正在扫描的字符所在位置，没有超过当前项目的MaxEndIndex



//
则我们需要对当前字符进行检查。



if
(i
<=
currentItem.MaxEndIndex)







{



//
是否为一个结束符？



FastDict dict
=
fastIndex[index];



if
(dict
!=

null
)







{



//
如果是，则从queue中的记录进行检索工作。



FastScanResult result
=
ScanWorkItems(text, i, dict, currentItem.LowPartDelta, queue);



//
找到了，自然是返回了。



if
(result
!=

null
)







{



return
result;



}



}





//
如果当前不是一个中间字符，同时也不是起始符，或者队列中有待处理的项目。



//
那么我们就需要清除待处理队列，避免不必要的重复计算。



if
((maxLength
==

0

||
queue.Count
!=

0
)
&&
(ctype
&
CharacterType.MiddleChar)
==
CharacterType.NoHit)







{





lastLowPart
=

0
;



queue.Clear();



currentItem.LowPartDelta
=

0
;



currentItem.MaxEndIndex
=
endIndex;



if
(endIndex
==

0
)







{



//
如果不是遇到起始符，则完全不需要尝试后面的计算。



continue
;



}



}





//
如果当前待处理项的结束位置已经超出当前位置，则需要尝试清理现有的工作项



if
(queue.Count
>

0

&&
currentItem.MaxEndIndex
<=
i)







{



//
下面的代码用于恢复最后待处理项哈希值的低16位



uint
lastLowPartDelta
=

0
;



uint
oldLowPart
=
currentItem.LowPartDelta;



do







{



currentItem
=
queue.Dequeue();



lastLowPartDelta
^=
currentItem.LowPartDelta;



}

while
(queue.Count
>

0

&&
currentItem.MaxEndIndex
<=
i);



currentItem.LowPartDelta
=
oldLowPart
^
lastLowPartDelta;



lastLowPart
^=
lastLowPartDelta;



//
只有当前位置仍然在最远可能结束符之前，我们就需要计算哈希值



if
(i
<
currentItem.MaxEndIndex)







{



currentItem.LowPartDelta
^=
index;



}



}



else







{



//
如果待处理项的结束位置没有超出当前位置，也需要计算哈希值



currentItem.LowPartDelta
^=
index;



}



}



}





//
如果循环到结束位置，仍然没有找到关键字，表示找不到了，此时返回空。



return

null
;



}



可以看到，上面这段代码确实是比普通的算法要复杂。提高效率自然是要付出代价的，有的时候我们可以通过多消耗点内存来达到，而更加根本、更加有效的
还是提高算法的“复杂度”。比如快速排序，其算法复杂度就比普通的冒泡、插入排序要高得多。即便是多消耗再多的内存，冒泡排序在处理速度上仍然是无法比得
过快速排序的。当然了，你可以说我的任务只需要每秒钟扫描3百万个字符，这个时候普通的算法就足以应付了。又或者例如我之前的情况，从晚上6点开始到早上
6点之前，必须要把所有的内容都检索一遍，于是没功夫想一个更好的算法，有什么代码只要能工作，写起来又简单就好。在这样的任务条件下，也许TTMP-F
算法是不适合的。而如果你现在已经有点时间，同时效率又是你最头痛的事情时，你还是不得不仔细看看上面的代码，况且我觉得还不算太复杂。

大部分的代码，我想大家还是轻易就能看明白的。关键有两个地方不太好理解：

1、为什么待处理项currentItem不放在队列中？"To do 1"是怎么实施的？

2、lastLowPart、LowPartDelta这些变量是什么意思？为什么要使用这些变量？

下面是部分的答案：

1、产生一个对象也许没有什么成本，但是如果产生上万次，累计起来的效率也是比较客观的。这还不是关键的，关键是整个核心就是和当前待处理的项目有关，如
果我们把当前待处理项加入到队列当中，则我们每次要对一些当前条件做判断的时候，是否也需要从队列中取出来？即使我们用Peek，其对于效率的影响也是非
常之高的，甚至该部分调用的次数，比产生currentItem的次数还要多。那么To do 1是怎么实施的呢？就算作我就留给大家的一个作业了。

2、如果大家看过我以前的文章，一定知道我这个人写东西有两个特点：一个是罗嗦，另一个就是爱吊人胃口。没错，放在下一集播出。其实不是我不想一口气写
完，是我发现由于废话比较多的缘故，写得太长了——断续写了好几天了，也没有一个结束。想到这个问题，还是觉得先发布一下吧，免得我一不小心IE崩溃，这
片心血就奔赴长江了……

那么，下一期的节目，还是会继续实战F模式。（还有好多代码没有贴出来哦……敬请期待）
http://www.cnblogs.com/sumtec/articles/1065407.aspx