转：C#实现简单分布式搜索--------希望有时间能好好研究一下

蛙蛙推荐：用c#实现一个简单的分布式搜索

[收藏此页] [打印]

【IT168知识库】

目标：输入一个关键字，从不同的资源库里检索出符合条件的资源条目。其中，资源库有本地硬盘上的数据，有远程web上的数据，其中前一种资源搜索由应用程序LocalSearcher来做，后一种资源的搜索由RemWebSearcher来做，而搜索的入口是一个网站DSearchWeb。DSearchWeb收到搜索请求后，分别起两个线程去调用LocalSearcher和RemWebSearcher，等它们两个的执行结果都回来后把结果组合到一起显示给客户，其中由于RemWebSearcher工作压力比较大，他和DSearchWeb不在一台机器上，它们之间靠Rmoting通信。实际应用中RemWebSearcher可以有多台来均衡搜索的压力，并且如果某台服务器搜索超时或者抛出异常，主入口程序不能崩溃。还有就是我们讲分布式搜索的结果拿到手后需要对结果进行一些排序或者敏感词过滤的操作。

其中本地搜索和远程搜索都只写了一些示例代码，本来应该用WebRequest和lucene来完善这块儿代码的，后来觉得和本文主题离的有些远就省略了先。

先来定义相关接口和公共类

/// <summary>
/// 搜索结果类，为了可以用Remoting传输，标记为可序列化
/// </summary>
[Serializable]
public class SearchResult
{
public SearchResult(string rst)
{
result = rst;
}
string result;

public string Result
{
get { return result; }
set { result = value; }
}
}
/// <summary>
/// 搜索器接口
/// </summary>
public interface ISearcher
{
List<SearchResult> Search(string keyword);
}
然后准备一个本地的搜索器和远程的搜索器

/// <summary>
/// 本地搜索器
/// </summary>
public class LocalSearcher :ISearcher
{
#region ISearcher 成员

public List<SearchResult> Search(string keyword)
{
//模拟用lucene搜索本地索引
List<SearchResult> list = new List<SearchResult>();
list.Add(new SearchResult("秋天不会来"));
// Thread.Sleep(1000);
list.Add(new SearchResult("风烟滚滚唱英雄"));
//Thread.Sleep(1000);
list.Add(new SearchResult("红藕香残玉簟秋"));
return list;
}
#endregion
}

远程搜索服务器为一个控制台程序，下面是其主要类

/// <summary>
/// 远程搜索类，为了让remoting暴露，继承MarshalByRefObject
/// </summary>
public class RemWebSearcher : MarshalByRefObject,ISearcher
{
#region ISearcher 成员
public List<SearchResult> Search(string keyword)
{
//模拟用webrequest来抓取网页，并找出符合条件的结果条目
List<SearchResult> list = new List<SearchResult>();
list.Add(new SearchResult("我是一条小青龙，小青龙"));
//Thread.Sleep(1000);
list.Add(new SearchResult("相见时难别亦难，东风无力白花残。"));
//Thread.Sleep(1000);
list.Add(new SearchResult("为什么命名相爱，到最后还是要重来。"));
return list;
}
#endregion
}
/// <summary>
/// 启动本台搜索服务器，真实环境中远比这个要复杂
/// </summary>
class Program
{
static void Main(string[] args)
{
TcpChannel channel = new TcpChannel(8080);
ChannelServices.RegisterChannel(channel,false);
RemotingConfiguration.RegisterWellKnownServiceType(
typeof(RemWeb.RemWebSearcher),
"RemWebSearcher", WellKnownObjectMode.Singleton);
Console.WriteLine("服务已启动,回车后停止");
Console.Read();
channel.StopListening(null);
ChannelServices.UnregisterChannel(channel);
Console.WriteLine("服务已停止,回车后退出 ");
//这里要两个read()，一个不顶事。
Console.Read();
Console.Read();

}
}

接下来，我们做一个网站来让用户使用，我们做成ajax的


<form id="form1" runat="server">


<asp:ScriptManager ID="ScriptManager1" runat="server" />


<div>


<asp:TextBox ID="txtKeyword" runat="server"></asp:TextBox>


<asp:Button ID="btnSearch" runat="server" Text="搜索" OnClick="btnSearch_Click" />


<asp:UpdateProgress ID="UpdateProgress1" runat="server" AssociatedUpdatePanelID="UpdatePanel1">


<ProgressTemplate>


Please Wait



</ProgressTemplate>


</asp:UpdateProgress>


<br />


<asp:UpdatePanel ID="UpdatePanel1" runat="server" UpdateMode="Conditional">


<ContentTemplate>


<asp:Repeater ID="rptResults" runat="server">


<ItemTemplate>


<asp:Label runat="server" ID="Label1" Text='<%# Eval("Result") %>' /><br/>


</ItemTemplate>


</asp:Repeater>


</ContentTemplate>


<Triggers>


<asp:AsyncPostBackTrigger ControlID="btnSearch" EventName="Click" />


</Triggers>


</asp:UpdatePanel>


</div>


</form>



还有，我们要连接远程的搜索服务器

/// <summary>
/// 在应用程序启动时创建到各搜索服务器的连接，关闭是注销相关通道
/// </summary>
public class Global : System.Web.HttpApplication
{

TcpChannel channel;
protected void Application_Start(object sender, EventArgs e)
{
channel = new TcpChannel();
ChannelServices.RegisterChannel(channel,false);

}

protected void Application_End(object sender, EventArgs e)
{
channel.StopListening(null);
ChannelServices.UnregisterChannel(channel);
}
}

剩下的就是分布式搜索的逻辑了




/**//// <summary>


/// 状态对象


/// </summary>


public class StateObj






{


public StateObj( ISearcher searcher,


string keyWord,


AutoResetEvent are)






{


Searcher = searcher;


KeyWord = keyWord;


AutoEvent = are;


}


public AutoResetEvent AutoEvent;


public string KeyWord;


public ISearcher Searcher;


}




public partial class _Default : System.Web.UI.Page






{


protected void Page_Load(object sender, EventArgs e)






{




}


List<SearchResult> m_list = new List<SearchResult>(); //全局搜索结果列表


Int32 numAsyncOps = 2; //异步操作的数量


AutoResetEvent areDoSearch = new AutoResetEvent(false); //全部搜索完成的事件


AutoResetEvent areDoFilter = new AutoResetEvent(false); //敏感此过滤完成的事件




//处理搜索按钮的方法


protected void btnSearch_Click(object sender, EventArgs e)






{


getSearchResults(txtKeyword.Text.Trim());


rptResults.DataSource = m_list;


rptResults.DataBind();


}






/**//// <summary>


/// 进行多线程和分布式搜索


/// </summary>


/// <param name="keyWord"></param>


private void getSearchResults(string keyWord)






{


//注册全部搜索完成后将执行的回调，回调里将结果进行敏感词过滤


//这里要设置一个合理的超时值，以防止有些服务器操作超时造成整体无响应


RegisteredWaitHandle rwh = ThreadPool.RegisterWaitForSingleObject(


areDoSearch,


doFiterResult,


null,


10000,


false);






创建搜索服务器列表#region 创建搜索服务器列表


ISearcher[] arrSearcher;


try






{


//这里用一个子方法来完成所有SearchServer的创建


//实际应用中应该用单件模式来实现，而不是每次搜索


//都创建一遍


arrSearcher = getSearcherList();




}


catch (Exception ex)






{


Trace.Warn("访问搜索服务器出错：" + ex.ToString());


return;


}


#endregion




numAsyncOps = arrSearcher.Length; //重置异步操作数量






调用线程池线程分别执行每个远程异步搜索#region 调用线程池线程分别执行每个远程异步搜索


AutoResetEvent[] ares = new AutoResetEvent[arrSearcher.Length];


int iSearcher = 0;


StateObj stateObj = null;


foreach (ISearcher searcher in arrSearcher)






{


//每个异步操作拥有一个自动事件，用来通知主线程完成还是超时


ares[iSearcher] = new AutoResetEvent(false);


stateObj = new StateObj(searcher, keyWord, ares[iSearcher]); //准备状态参数




try






{


if (!ThreadPool.QueueUserWorkItem(new WaitCallback(doSearch),


stateObj))






{


//如果无法压入线程池工作队列，记录下日志


Trace.Warn("Unable to queue ThreadPool request.");


}


}


catch (ApplicationException ae) //调用QueueUserWorkItem有可能会有异常






{


Trace.Warn("服务器内存不足" + ae.ToString());


}


iSearcher++;


}


#endregion






等待每一个异步搜索直到完成或者超时#region 等待每一个异步搜索直到完成或者超时


//这个等待主要是为了防止某台服务器超时，否则


//本循环的最后两句可以写到doSearch的最后


//但是那样如果有一台服务器长时间无响应，这个搜索过程就会无响应了


foreach (AutoResetEvent are in ares)






{


if (are.WaitOne(new TimeSpan(0, 0, 15), false))






{


//这里只记录某台，实际中应该记录详细的机器编号


Trace.Warn("某台搜索服务器执行成功");


}


else






{


Trace.Warn("某台搜索服务器执行超时");


}


//如果全部异步搜索已完成或超时，那么设置areDoSearch为阻止状态


//此时会触发doFiterResult方法去对结果进行敏感词过滤


//Interlocked保证递增和递减为一个CPU的原子操作


if (Interlocked.Decrement(ref numAsyncOps) == 0)


areDoSearch.Set();


}


#endregion




areDoFilter.WaitOne(); //等待敏感词过滤完毕


rwh.Unregister(null); //注销areDoSearch事件


}






/**//// <summary>


/// 创建搜索服务器列表，真实情况下建议使用工厂方法模式


/// </summary>


/// <returns></returns>


private ISearcher[] getSearcherList()






{


List<ISearcher> list= new List<ISearcher>();


list.Add(new LocalSearcher());




ISearcher searcher = (ISearcher)Activator.GetObject(typeof(ISearcher),


"tcp://localhost:8080/RemWebSearcher");


list.Add(searcher);


return list.ToArray();


}






/**//// <summary>


/// 进行结果过滤操作，真实情况下需要进行关联度分析后排序，以及敏感词过滤等操作


/// </summary>


/// <param name="state"></param>


/// <param name="timeOut"></param>


void doFiterResult(Object state, bool timeOut)






{


Trace.Warn("{" + Thread.CurrentThread.GetHashCode().ToString() + "}:doFiterResult

");


if (timeOut)






{


Trace.Warn("搜索超时，某些搜索服务器可能没有在指定的时间内返回结果");


}


lock (((ICollection)m_list).SyncRoot)






{


//这里模拟一些敏感词过滤的逻辑


m_list.Add(new SearchResult("嘿嘿一定是你吧"));


}


areDoFilter.Set(); //设置敏感词过滤事件为终止，通知主线程搜索结果已经可用，可以展示给用户


}






/**//// <summary>


/// 每个搜索器的执行方法


/// </summary>


/// <param name="state"></param>


void doSearch(Object state)






{


Trace.Warn("{" + Thread.CurrentThread.GetHashCode().ToString() + "}:doSearch

");






处理该线程上的未处理异常，虽然对搜索过程加了cathc，但还是要防止漏网之鱼#region 处理该线程上的未处理异常，虽然对搜索过程加了cathc，但还是要防止漏网之鱼


AppDomain.CurrentDomain.UnhandledException += delegate(object sender,


UnhandledExceptionEventArgs e)






{


Trace.Warn(Thread.CurrentThread.GetHashCode().ToString()


+ "线程上有未处理的异常:" +


((Exception)e.ExceptionObject).ToString());


};


#endregion






StateObj stateObj = (StateObj)state;




ISearcher searcher = stateObj.Searcher; //当前的搜索器


string keyWord = stateObj.KeyWord; //搜索的关键词






执行搜索器的搜索方法获取搜索结果，并捕获异常#region 执行搜索器的搜索方法获取搜索结果，并捕获异常


IList<SearchResult> lst = null;


try






{


lst = searcher.Search(keyWord);


}


catch (Exception ex)






{


Trace.Warn("doSearch has an error:" + ex.ToString());


}


#endregion








拿到结果后，把结果合并到全局搜索结果列表里#region 拿到结果后，把结果合并到全局搜索结果列表里


//考虑到多线程访问全局结果列表，这里要对其加锁


if (null != lst)






{


lock (((ICollection)m_list).SyncRoot)






{


m_list.AddRange(lst);


}


}


#endregion




stateObj.AutoEvent.Set(); //通知主线程，本搜索器执行完毕，可以等下一个了。


}




}





相关链接：
NET多线程同步和互斥机制初探
http://www.cnblogs.com/herony420/articles/221523.html
使用线程池
http://www.cnblogs.com/three/archive/2006/10/11/526082.html
如何对制造者线程和使用者线程进行同步
http://www.cnblogs.com/three/archive/2006/09/29/518519.html
如何：创建和终止线程
http://www.cnblogs.com/three/archive/2006/09/30/519034.html
HOW TO：使用 Visual C# .NET 同步对多线程环境中共享资源的访问
http://support.microsoft.com/kb/816161/zh-cn
@@@@C#线程池的并发问题，高手进@@@@
http://www.ttscj.cn/info/60172.htm
Visual C#中的多线程编程
http://www.manbu.net/article.asp?id=39
WaitHandle.WaitOne Method (TimeSpan, Boolean)
http://msdn2.microsoft.com/en-us/library/85bbbxt9.aspx
Microsoft .Net Remoting系列专题之一:.Net Remoting基础篇
/article/4591721.html

第一次写多线程的帖子，也是初学，可能有好多考虑不全的地方，还请各位老大指正。
相关图片和代码我就贴了。
刚开始贴的代码有些问题，不好意思。

源码下载地址如下，大家自己下载调试吧。

http://www.cnblogs.com/Files/onlytiancai/DSearch.zip