在SQL Server实现最短路径的搜索

star870126

　　

开始

　　
　　这是去年的问题了，今天在整理邮件的时候才发现这个问题，感觉顶有意思的，特记录下来。
　　在表RelationGraph中，有三个字段（ID,Node,RelatedNode）,其中Node和RelatedNode两个字段描述两个节点的连接关系；现在要求，找出从节点"p"至节点"j"，最短路径（即经过的节点最少）。

　　图1.
　　
　　

解析

　　
　　为了能够更好的描述表RelationGraph中字段Node和 RelatedNode的关系，我在这里特意使用一个图形来描述，如图2.

　　图2.
　　
　　在图2，可清晰的看出各个节点直接如何相连，也可以清楚的看出节点"p"至节点"j"的的几种可能路径。

　　从上面可以看出第2种可能路径，经过的节点最少。
　　为了解决开始的问题，我参考了两种方法，
　　第1方法是，
　　参考单源最短路径算法：Dijkstra(迪杰斯特拉)算法，主要特点是以起始点为中心向外层层扩展，直到扩展到终点为止。

　　图3.
　　第2方法是，
　　针对第1种方法的改进，就是采用多源点方法，这里就是以节点"p"和节点"j"为中心向外层扩展，直到两圆外切点，如图4. ：

　　图4.
　　
　　

实现

　　在接下来，我就描述在SQL Server中，如何实现。当然我这里采用的前面说的第2种方法，以"P"和"J"为始点像中心外层层扩展。
　　（注：以下的脚本是在SQL Server 2012测试通过，也可运行在SQL Server 2008/2008R2上）
　　这里提供有表RelactionGraph的create& Insert数据的脚本：

use TestDB      　　go 　　 　　if object_id('RelactionGraph') Is not null drop table RelactionGraph 　　create table RelactionGraph(ID int identity,Item nvarchar(50),RelactionItem nvarchar(20),constraint PK_RelactionGraph primary key(ID)) 　　go 　　create nonclustered index IX_RelactionGraph_Item on RelactionGraph(Item) include(RelactionItem) 　　create nonclustered index IX_RelactionGraph_RelactionItem on RelactionGraph(RelactionItem) include(Item) 　　go 　　 　　insert into RelactionGraph (Item, RelactionItem ) values 　　    ('a','b'),('a','c'),('a','d'),('a','e'), 　　    ('b','f'),('b','g'),('b','h'), 　　    ('c','i'),('c','j'), 　　    ('f','k'),('f','l'), 　　    ('k','o'),('k','p'), 　　    ('o','i'),('o','l') 　　 　　go

　　编写一个存储过程up_GetPath

useTestDB 　　go 　　--Procedure: 　　ifobject_id('up_GetPath')Isnotnull 　　    Dropprocup_GetPath 　　go 　　createprocup_GetPath 　　( 　　    @Nodenvarchar(50), 　　    @RelatedNodenvarchar(50) 　　) 　　As 　　setnocounton 　　 　　declare 　　    @levelsmallint=1,--当前搜索的深度 　　    @MaxLevelsmallint=100,--最大可搜索深度 　　    @Node_WhileFlagbit=1,--以@Node作为中心进行搜索时候，作为能否循环搜索的标记 　　    @RelatedNode_WhileFlagbit=1 --以@RelatedNode作为中心进行搜索时候，作为能否循环搜索的标记 　　 　　--如果直接找到两个Node存在直接关系就直接返回 　　ifExists(select 1 fromRelationGraphwhere (Node=@Node And RelatedNode=@RelatedNode)or(Node=@RelatedNode And RelatedNode=@Node))or @Node=@RelatedNode 　　begin 　　    selectconvert(nvarchar(2000),@Node +' --> '+@RelatedNode)As RelationGraphPath,convert(smallint,0) AsStopCount 　　    return 　　end 　　 　　-- 　　 　　ifobject_id('tempdb..#1')IsnotnullDropTable#1--临时表#1，存储的是以@Node作为中心向外扩展的各节点数据 　　ifobject_id('tempdb..#2')IsnotnullDropTable#2--临时表#2，存储的是以@RelatedNode作为中心向外扩展的各节点数据 　　 　　createtable#1( 　　    Nodenvarchar(50),--相对源点 　　    RelatedNodenvarchar(50),--相对目标 　　    Levelsmallint--深度 　　    ) 　　    　　createtable#2(Nodenvarchar(50),RelatedNodenvarchar(50),Levelsmallint) 　　 　　insertinto#1(Node, RelatedNode,Level) 　　    selectNode, RelatedNode, @levelfromRelationGraphawherea.Node =@Nodeunion--正向：以@Node作为源查询 　　    selectRelatedNode, Node, @levelfromRelationGraphawherea.RelatedNode = @Node--反向：以@Node作为目标进行查询 　　set@Node_WhileFlag=sign(@@rowcount) 　　    　　insertinto#2(Node, RelatedNode,Level) 　　    selectNode, RelatedNode, @levelfromRelationGraphawherea.Node =@RelatedNodeunion--正向：以@RelatedNode作为源查询 　　    selectRelatedNode, Node, @levelfromRelationGraphawherea.RelatedNode = @RelatedNode--反向：以@RelatedNode作为目标进行查询 　　set@RelatedNode_WhileFlag=sign(@@rowcount) 　　 　　--如果在表RelationGraph中找不到@Node 或 @RelatedNode 数据，就直接跳过后面的While过程 　　ifnotexists(select 1 from#1)ornotexists(select 1 from#2) 　　begin 　　    gotoWhile_Out 　　end 　　 　　 　　whilenotexists(select 1 from#1a innerjoin#2bonb.RelatedNode=a.RelatedNode)--判断是否出现切点 　　     and(@Node_WhileFlag|@RelatedNode_WhileFlag)>0 --判断是否能搜索 　　     And@level0 　　    begin    　　        insertinto#1(Node, RelatedNode,Level) 　　            --正向 　　            selecta.Node,a.RelatedNode,@level+1 　　                FromRelationGrapha 　　                whereexists(select 1 from#1whereRelatedNode=a.Node AndLevel=@level)And 　　                    Notexists(select 1 from#1whereNode=a.Node)             　　            union 　　            --反向 　　            selecta.RelatedNode,a.Node,@level+1 　　                FromRelationGrapha 　　                whereexists(select 1 from#1whereRelatedNode=a.RelatedNode AndLevel=@level)And 　　                    Notexists(select 1 from#1whereNode=a.RelatedNode) 　　         　　        set@Node_WhileFlag=sign(@@rowcount) 　　 　　    end 　　 　　    　　    if@RelatedNode_WhileFlag>0 　　    begin         　　        insertinto#2(Node, RelatedNode,Level) 　　            --正向 　　            selecta.Node,a.RelatedNode,@level+1 　　                FromRelationGrapha 　　                whereexists(select 1 from#2whereRelatedNode=a.Node AndLevel=@level)And 　　                    Notexists(select 1 from#2whereNode=a.Node) 　　            union 　　            --反向 　　            selecta.RelatedNode,a.Node,@level+1 　　                FromRelationGrapha 　　                whereexists(select 1 from#2whereRelatedNode=a.RelatedNode AndLevel=@level)And 　　                    Notexists(select 1 from#2whereNode=a.RelatedNode) 　　        set@RelatedNode_WhileFlag=sign(@@rowcount) 　　    end 　　    　　    select@level+=1 　　end 　　 　　While_Out: 　　 　　--下面是构造返回的结果路径 　　ifobject_id('tempdb..#Path1')IsnotnullDropTable#Path1 　　ifobject_id('tempdb..#Path2')IsnotnullDropTable#Path2 　　 　　;withcte_path1As 　　( 　　selecta.Node,a.RelatedNode,Level,convert(nvarchar(2000),a.Node+' -> '+a.RelatedNode)As RelationGraphPath,Convert(smallint,1) AsPathLevelFrom#1awhere exists(select 1 from#2whereRelatedNode=a.RelatedNode) 　　unionall 　　selectb.Node,a.RelatedNode,b.Level,convert(nvarchar(2000),b.Node+' -> '+a.RelationGraphPath)As RelationGraphPath ,Convert(smallint,a.PathLevel+1)As PathLevel 　　    fromcte_path1a 　　        innerjoin#1bonb.RelatedNode=a.Node 　　            andb.Level=a.Level-1 　　) 　　select*Into#Path1fromcte_path1 　　 　　;withcte_path2As 　　( 　　selecta.Node,a.RelatedNode,Level,convert(nvarchar(2000),a.Node)As RelationGraphPath,Convert(smallint,1) AsPathLevelFrom#2awhere exists(select 1 from#1whereRelatedNode=a.RelatedNode) 　　unionall 　　selectb.Node,a.RelatedNode,b.Level,convert(nvarchar(2000),a.RelationGraphPath+' -> '+b.Node)As RelationGraphPath ,Convert(smallint,a.PathLevel+1) 　　    fromcte_path2a 　　        innerjoin#2bonb.RelatedNode=a.Node 　　            andb.Level=a.Level-1 　　) 　　select*Into#Path2fromcte_path2 　　 　　;withcte_resultAs 　　( 　　selecta.RelationGraphPath+' -> '+b.RelationGraphPathAsRelationGraphPath,a.PathLevel+b.PathLevel -1 As StopCount,rank()over(orderbya.PathLevel+b.PathLevel)As Result_row 　　    From#Path1a 　　        innerjoin#Path2bonb.RelatedNode=a.RelatedNode 　　            andb.Level=1 　　    wherea.Level=1 　　)    　　selectdistinctRelationGraphPath,StopCountFromcte_resultwhereResult_row=1 　　go

　　
　　上面的存储过程，主要分为两大部分，第1部分是实现如何搜索，第2部分实现如何构造返回结果。其中第1部分的代码根据前面的方法2，通过@Node 和 @RelatedNode 两个节点向外层搜索，每次搜索返回的节点都保存至临时表#1和#2，再判断临时表#1和#2有没有出现切点，如果出现就说明已找到最短的路径（经过多节点数最少），否则就继续循环搜索，直到循环至最大的搜索深度（@MaxLevel smallint=100）或找到切点。要是到100层都没搜索到切点，将放弃搜索。这里使用最大可搜索深度@MaxLevel，目的是控制由于数据量大可能会导致性能差，因为在这里数据量与搜索性能成反比。代码中还说到一个正向和反向搜索，主要是相对Node 和 RelatedNode来说，它们两者互为参照对象，进行向外搜索使用。
　　下面是存储过程的执行：

useTestDB 　　go 　　execdbo.up_GetPath 　　        @Node='p', 　　@RelatedNode='j' 　　 　　go

　　

　　你可以根据需要来，赋予@Node 和 @RelatedNode不同的值。
　　

扩展

　　前面的例子，可扩展至城市的公交路线，提供两个站点，搜索经过这两个站点最少站点公交路线；可以扩展至社区的人际关系的搜索，如一个人与另一个人想认识，那么他们直接要经过多少个人才可以。除了人与人直接有直接的朋友、亲戚关联，还可以通过人与物有关联找到人与人关联，如几个作家通过出版一个本，那么就说明这几个人可以通过某一本书的作者列表中找到他们存在共同出版书籍的关联，这为搜索两个人认识路径提供参考。这问题可能会非常大复杂，但可以这样的扩展。
　　

小结

　　
　　这里只是找两个节点的所有路径中，节点数最少的路径，在实际的应用中，可能会碰到比这里更复杂的情况。在其他的环境或场景可能会带有长度，时间，多节点，多作用域等一些信息。无论如何，一般都要参考一些原理，算法来实现。