又想速度快，又要大数据，又要保证数据不出错，还要拥抱变化，改需求的时候不那么痛苦，特别是字段的调整，按照以前的做法，想想就头疼。使用NoSQL，简直就是随心所欲，再奇葩的数据结构，处理起来也很容易。下面看我如何用NoSQL数据库实现高并发，高可靠的CRM系统。
1、前言
随着facebook、微博等WEB2.0互联网网站的兴起，传统的关系数据库在应付web2.0网站，特别是超大规模和高并发的SNS类型的web2.0纯动态网站已经显得力不从心，暴露了很多难以克服的问题，而非关系型的数据库则由于其本身的特点得到了非常迅速的发展。
NoSQL项目的名字上看不出什么相同之处，但是，它们通常在某些方面相同：它们可以处理超大量的数据。
目前NoSQL数据库的发展非常迅速，各大公司都在大量使用，比如GOOGLE、360、百度，在不久的将来必将成为主流。
为了适应大数据处理的需要，在将来较长的一段时间里能够让系统适应高并发的访问，我们在开发CRM系统的过程中，经过不断摸索和实践，通过使用NoSQL数据库，解决了传统关系型数据很难解决的难题，实现了大数据和高并发的处理能力。
2、系统展示
3、性能数据
测试环境是CentOS系统，ngnix + php5.0， i7 2600K处理器，8G内存，SSDB 1.6.8.8版本。
针对最常用的功能，以前数据量最大的功能进行测试。
用PHP程序模拟UI层直接调用业务逻辑接口，插入大量数据，并对大量数据进行读取。
功能 |
测试说明 |
测试结果（毫秒） |
添加客户资料 |
添加100万条数据 |
15203 |
客户资料查询 |
对100万条客户资料进行查询1万次 |
20 |
下订单 |
添加100万条数据 |
58653 |
订单查询 |
对100万条订单进行查询1万次 |
25 |
添加新消息 |
添加100万条数据 |
13454 |
获取新消息 |
对100万条消息数据进行查询1万次 |
18 |
4、系统架构
在使用数据库的选择上，我们使用同时具有高性能，并且接口使用很简单的SSDB。因为SSDB的作者是我的朋友，同时SSDB已经在百度、360等大公司大量应用，所以不用担心可靠性问题。
开发语言使用PHP。
4.1 系统需求
2014年禾葡兰借着微信营销的大好机会，在几个月时间里，团队发展到了100人以上，每日订单数超过200单。
在快速发展的过程中，对于数据处理的需求日益增长，通过数据和统计进行决策，使经营水平和收入更上一个台阶，并能够适应将来的大规模扩展的需要。
目前纯手工统计和计算比较容易出错，而且效率很低，很难适应快速增长的需要，为了解决这些问题，欣所罗门核心团队成员一致认为有必要开发一个软件系统来解决问题。
4.2 需要解决的3大难点
1、事务处理
在数据完整性要求很高的CRM系统中，必需要事务处理机制，保证原子操作。但是，NoSQL数据库最大的缺点就是不支持事务处理。
要想实现事务处理，目前最好的方法是使用zookeeper，可以模拟实现分布式事务。
2、数据关系
和关系数据库不同的是，SSDB没有关系的概念。所有数据间的关系必需在开始的时候设计好。
3、组合条件查询
由于NoSQL数据库没有模式，因此，想要根据数据内容里的某个字段对数据集进行筛选，根本不可能。所以，我们使用lucene实现全文索引，解决了组合条件查询的问题。
4.3 系统设计
由于目前流行的PHP开发框架MVC框架都不支持NoSQL数据，同时避免由于开源框架带来的安全问题，因此不使用MVC框架。仅在UI层使用php的smarty模板引擎。
系统统一入口文件index.php，通过m参数引用ui目录和controller目录下相应的文件。
代码目录结构
/ CRM系统根目录
/index.php (统一入口文件)
|-------common/ (用于存放公共函数)
|--------common.php (公共函数接口文件，使用公共函数只需要引用一个文件就可以。)
|-------config/ (用于存放配置信息)
|-------model/ (数据对像)
|-------data/ (数据层，对数据的读写封装。)
|-------controller/ (业务逻辑层)
|-------ui/ (smarty引擎和模板)
|-------libs/ (smarty引擎目录)
|-------configs/ (smarty引擎配置文件目录)
|-------plugins/ (smarty引擎自定义的一些实用插件目录)
|-------templates/ (smarty引擎模板目录)
|-------templates_c/ (smarty引擎模板编译目录)
|-------cache/ (smarty缓存目录)
5、涉及内容
一、 数据库连接
二、 自动编号ID实现
三、数据的读写
四、 数据关系实现
五、 分布式事务实现
六、 全文索引
七、 组合条件查询
6、源码解析
一、数据库连接
SSDB的数据库连接很简单。
include_once('SSDB.php');
try{
$ssdb=newSimpleSSDB('127.0.0.1',8888);
}catch(SSDBException$e){
die(__LINE__.''.$e->getMessage());
}
include_once('SSDB.php');
try{
$ssdb=newSimpleSSDB('127.0.0.1',8888);
}catch(SSDBException$e){
die(__LINE__.''.$e->getMessage());
}
二、 自动编号ID实现
一般来说，我们在进行数据设计的时候，都会给每条记录设置一个自动编号的ID。但是在NoSQL数据中，需要自己来实现自动编号ID。
我们通过SSDB的incr接口，就可以模拟自动编号ID的实现。
系统初始化的时候，我们在数据库中增加一条数据。
$ssdb->set('hpl_product_autoincrement_id',0);
$ssdb->set('hpl_product_autoincrement_id',0);
添加新记录的时候，使用incr接口得到新的自动编号ID，由于incr接口是原子操作，所以不会出现重复的ID。
$id=$ssdb->incr('hpl_product_autoincrement_id',1);
$id=$ssdb->incr('hpl_product_autoincrement_id',1);
三、 数据的读写
首先，我们要创建数据的model类。下面是产品model类。
classProductModel{
public$id;//编号
public$catalogid;//所属分类
public$name;//产品名称
public$cost;//产品成本价格
public$price;//销售价格
public$saleprice;//优惠价格
public$amount;//产品数量
public$facetype;//适合皮肤类型
public$desc;//产品描述
public$code;//产品编号
public$weight;//净含量（克）
public$effect;//主要功效
public$crowd;//适合人群
public$addtime;//产品添加时间
public$status;//产品状态（0=下架1=上架)
}
classProductModel{
public$id; //编号
public$catalogid; //所属分类
public$name; //产品名称
public$cost; //产品成本价格
public$price; //销售价格
public$saleprice; //优惠价格
public$amount; //产品数量
public$facetype; //适合皮肤类型
public$desc; //产品描述
public$code; //产品编号
public$weight; //净含量（克）
public$effect; //主要功效
public$crowd; //适合人群
public$addtime; //产品添加时间
public$status; //产品状态（0=下架1=上架)
}
新添加记录。
$product=newProductModel();
$product->id=$id;
...
$key='hpl_product_'.$id;
$json=json_encode($product);
$ssdb->hset('hpl_product',$key,$json);
$product=newProductModel();
$product->id=$id;
...
$key='hpl_product_'.$id;
$json=json_encode($product);
$ssdb->hset('hpl_product',$json);
增加索引，根据产品ID进行索引，在使用数据的时候就可以根据产品ID获取数据列表（如果有多种排列方式就需要创建多个索引）。
$key='hpl_product_'.$id;
$ssdb->zset('hpl_product_id',$id);
$key='hpl_product_'.$id;
$ssdb->zset('hpl_product_id',$id);
根据索引获取产品列表（前10个）。
$products=array();
//根据索引取出产品列表
$items=$ssdb->zscan('hpl_product_id','',10);
foreach($itemsas$key=>$score){
//取出产品信息
$json=$ssdb->hget('hpl_product',$key);
$products[]=json_decode($json);
}
四、数据关系实现
在添加产品的时候，需要把产品和产品分类关联起来，可以根据产品分类列出产品。使用SSDB我们只需要对每个产品分类创建一个列表就可以了。
添加关系，在列表里添加一条数据。
$hname='hpl_product_catalog_'.$catalogid; $hkey='hpl_product_'.$id; $ssdb->hset($hname,$hkey,$id);
$hname='hpl_product_catalog_'.$catalogid; $hkey='hpl_product_'.$id; $ssdb->hset($hname,$id);
删除关系，当改变产品分类或者删除产品的时候执行。
$hname='hpl_product_catalog_'.$catalogid; $hkey='hpl_product_'.$id; $ssdb->hdel($hname,$hkey);
$hname='hpl_product_catalog_'.$catalogid; $hkey='hpl_product_'.$id; $ssdb->hdel($hname,$hkey);
根据分类取出产品列表（前10个）。
$products=array();
//根据分类取出产品列表
$hname='hpl_product_catalog_'.$catalogid;
$keys=$ssdb->hkeys($hname,10);
foreach($keysas$key){
//取出产品信息
$json=$ssdb->hget('hpl_product',$key);
$products[]=json_decode($json);
}
$products=array(); //根据分类取出产品列表 $hname='hpl_product_catalog_'.$catalogid; $keys=$ssdb->hkeys($hname,$key); $products[]=json_decode($json); }
五、分布式事务实现
由于CRM系统对数据统一性的需要，所以必需要有事务来支持。比如在下订单的时候，如果没有事务，在并发处理时就有可能导致产品库存出错。由于SSDB数据库没有实现事务处理，我们使用zookeeper来实现分布式锁的处理，保证关键业务的原子操作。
事务处理的实现我们分别由前台和后台实现，前台系统发送业务处理请求，后台进程收到消息后进行处理，后台进程处理结束后设置处理标志，前台系统每隔一段时间查询一次标志位，检查业务是否处理完成。
后台实现使用JAVA语言编写。
/**
Executor.java
*/
importjava.io.FileOutputStream;
importjava.io.IOException;
importjava.io.InputStream;
importjava.io.OutputStream;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.ZooKeeper;
publicclassExecutor
implementsWatcher,Runnable,DataMonitor.DataMonitorListener
{
Stringznode;
DataMonitordm;
ZooKeeperzk;
Stringfilename;
Stringexec[];
Processchild;
publicExecutor(StringhostPort,Stringznode,Stringfilename,Stringexec[])throwsKeeperException,IOException{
this.filename=filename;
this.exec=exec;
zk=newZooKeeper(hostPort,3000,this);
dm=newDataMonitor(zk,znode,null,this);
}
/**
*@paramargs
*/
publicstaticvoidmain(String[]args){
if(args.length<4){
System.err
.println("USAGE:ExecutorhostPortznodefilenameprogram[args...]");
System.exit(2);
}
StringhostPort=args[0];
Stringznode=args[1];
Stringfilename=args[2];
Stringexec[]=newString[args.length-3];
System.arraycopy(args,3,exec,exec.length);
try{
newExecutor(hostPort,filename,exec).run();
}catch(Exceptione){
e.printStackTrace();
}
}
/***************************************************************************
*Wedoprocessanyeventsourselves,wejustneedtoforwardthemon.
*
*@seeorg.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
*/
publicvoidprocess(WatchedEventevent){
dm.process(event);
}
publicvoidrun(){
try{
synchronized(this){
while(!dm.dead){
wait();
}
}
}catch(InterruptedExceptione){
}
}
publicvoidclosing(intrc){
synchronized(this){
notifyAll();
}
}
staticclassStreamWriterextendsThread{
OutputStreamos;
InputStreamis;
StreamWriter(InputStreamis,OutputStreamos){
this.is=is;
this.os=os;
start();
}
publicvoidrun(){
byteb[]=newbyte[80];
intrc;
try{
while((rc=is.read(b))>0){
os.write(b,rc);
}
}catch(IOExceptione){
}
}
}
publicvoidexists(byte[]data){
if(data==null){
if(child!=null){
System.out.println("Killingprocess");
child.destroy();
try{
child.waitFor();
}catch(InterruptedExceptione){
}
}
child=null;
}else{
if(child!=null){
System.out.println("Stoppingchild");
child.destroy();
try{
child.waitFor();
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
try{
FileOutputStreamfos=newFileOutputStream(filename);
fos.write(data);
fos.close();
}catch(IOExceptione){
e.printStackTrace();
}
try{
System.out.println("Startingchild");
child=Runtime.getRuntime().exec(exec);
newStreamWriter(child.getInputStream(),System.out);
newStreamWriter(child.getErrorStream(),System.err);
}catch(IOExceptione){
e.printStackTrace();
}
}
}
}
/**
DataMonitor.java
*/
importjava.util.Arrays;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.ZooKeeper;
importorg.apache.zookeeper.AsyncCallback.StatCallback;
importorg.apache.zookeeper.KeeperException.Code;
importorg.apache.zookeeper.data.Stat;
importcom.udpwork.ssdb.SSDB;
importcom.udpwork.ssdb.Link;
importcom.udpwork.ssdb.MemoryStream;
importcom.udpwork.ssdb.Response;
publicclassDataMonitorimplementsWatcher,StatCallback{
ZooKeeperzk;
Stringznode;
WatcherchainedWatcher;
booleandead;
DataMonitorListenerlistener;
byteprevData[];
publicDataMonitor(ZooKeeperzk,WatcherchainedWatcher,DataMonitorListenerlistener){
this.zk=zk;
this.znode=znode;
this.chainedWatcher=chainedWatcher;
this.listener=listener;
//Getthingsstartedbycheckingifthenodeexists.Wearegoing
//tobecompletelyeventdriven
zk.exists(znode,true,this,null);
}
/**
*OtherclassesusetheDataMonitorbyimplementingthismethod
*/
publicinterfaceDataMonitorListener{
/**
*Theexistencestatusofthenodehaschanged.
*/
voidexists(bytedata[]);
/**
*TheZooKeepersessionisnolongervalid.
*
*@paramrc
*theZooKeeperreasoncode
*/
voidclosing(intrc);
}
publicvoidprocess(WatchedEventevent){
Stringpath=event.getPath();
if(event.getType()==Event.EventType.None){
//Wearearebeingtoldthatthestateofthe
//connectionhaschanged
switch(event.getState()){
caseSyncConnected:
//Inthisparticularexamplewedon'tneedtodoanything
//here-watchesareautomaticallyre-registeredwith
//serverandanywatchestriggeredwhiletheclientwas
//disconnectedwillbedelivered(inorderofcourse)
break;
caseExpired:
//It'sallover
dead=true;
listener.closing(KeeperException.Code.SessionExpired);
break;
}
}else{
if(path!=null&&path.equals(znode)){
//Somethinghaschangedonthenode,let'sfindout
//读取订单信息
//znode=/product/order/
...
//计算库存是否充足
...
//生成订单数据
...
//减少产品库存
...
zk.exists(znode,null);
}
}
if(chainedWatcher!=null){
chainedWatcher.process(event);
}
}
publicvoidprocessResult(intrc,Stringpath,Objectctx,Statstat){
booleanexists;
switch(rc){
caseCode.Ok:
exists=true;
break;
caseCode.NoNode:
exists=false;
break;
caseCode.SessionExpired:
caseCode.NoAuth:
dead=true;
listener.closing(rc);
return;
default:
//Retryerrors
zk.exists(znode,null);
return;
}
byteb[]=null;
if(exists){
try{
b=zk.getData(znode,false,null);
}catch(KeeperExceptione){
//Wedon'tneedtoworryaboutrecoveringnow.Thewatch
//callbackswillkickoffanyexceptionhandling
e.printStackTrace();
}catch(InterruptedExceptione){
return;
}
}
if((b==null&&b!=prevData)
||(b!=null&&!Arrays.equals(prevData,b))){
listener.exists(b);
prevData=b;
}
}
}
php zookeeper处理类
classOrderWorkerextendsZookeeper{
constCONTAINER='/product/order';
protected$acl=array(
array(
'perms'=>Zookeeper::PERM_ALL,'scheme'=>'world','id'=>'anyone'));</p><p>private$znode;
publicfunction__construct($host='',$watcher_cb=null,$recv_timeout=10000){
parent::__construct($host,$watcher_cb,$recv_timeout);
}
//添加订单
publicfunctionAdd($order){
if(!$this->exists(self::CONTAINER)){
$this->create(self::CONTAINER,$this->acl);
}
$this->znode=$this->create(self::CONTAINER.'/w-',$this->acl,Zookeeper::EPHEMERAL|Zookeeper::SEQUENCE);
$this->set($this->znode,json_encode(array('status'=>0,'data'=>$order,'orderid'=>'')));
$this->znode=str_replace(self::CONTAINER.'/',$this->znode);
return$this->znode;
}
//获取订单处理信息
publicfunctionGet($znode){
$data=$this->get(self::CONTAINER.'/'.$znode);
$json=json_decode($data);
return$json;
}
//删除订单znode
publicfunctionDel($znode){
$this->delete(self::CONTAINER.'/'.$znode);
}
}
classOrderWorkerextendsZookeeper{
constCONTAINER='/product/order';
protected$acl=array(
array(
'perms'=>Zookeeper::PERM_ALL,$this->znode);
return$this->znode;
}
//获取订单处理信息
publicfunctionGet($znode){
$data=$this->get(self::CONTAINER.'/'.$znode);
$json=json_decode($data);
return$json;
}
//删除订单znode
publicfunctionDel($znode){
$this->delete(self::CONTAINER.'/'.$znode);
}
}
PHP前台下订单：
$worker=newOrderWorker('127.0.0.1:2181');
$znode=$worker->Add($order);
echojson_encode(array('code'=>0,'znode'=>$znode));
echojson_encode(array('code'=>0,'znode'=>$znode));
PHP前台检测订单：
$worker=newOrderWorker('127.0.0.1:2181');
$data=$worker->Get($znode);
if(intval($data->status)==1)
{
$worker->Del($znode);
echojson_encode(array('status'=>1,'msg'=>'订单处理成功','orderid'=>$data->orderid));
}
elseif(intval($data->status)==2)
{
$worker->Del($znode);
echojson_encode(array('status'=>2,'msg'=>'订单处理失败'));
}
else
{
echojson_encode(array('status'=>0,'msg'=>'正在处理'));
}
$worker=newOrderWorker('127.0.0.1:2181');
$data=$worker->Get($znode);
if(intval($data->status)==1)
{
$worker->Del($znode);
echojson_encode(array('status'=>1,'orderid'=>$data->orderid));
}
elseif(intval($data->status)==2)
{
$worker->Del($znode);
echojson_encode(array('status'=>2,'msg'=>'订单处理失败'));
}
else
{
echojson_encode(array('status'=>0,'msg'=>'正在处理'));
}
六、全文索引
创建索引
importjava.io.File;
importjava.io.FileReader;
importjava.io.IOException;
importorg.apache.lucene.analysis.standard.StandardAnalyzer;
importorg.apache.lucene.document.Document;
importorg.apache.lucene.document.Field;
importorg.apache.lucene.document.LongField;
importorg.apache.lucene.document.StringField;
importorg.apache.lucene.document.TextField;
importorg.apache.lucene.index.IndexWriter;
importorg.apache.lucene.index.IndexWriterConfig;
importorg.apache.lucene.store.Directory;
importorg.apache.lucene.store.FSDirectory;
importorg.apache.lucene.util.Version;
publicclassIndexOrders{
privateIndexWriterwriter=null;
publicvoidAdd(StringindexPath,OrderModelorder)
throwsIOException{
//获取放置索引文件的位置，若传入参数为空，则读取search.properties中设置的默认值。
if(indexPath==null){
indexPath=LoadProperties.getProperties("indexDir");
}
finalFileindexDir=newFile(indexPath);
if(!indexDir.exists()||!indexDir.canRead()){
System.out
.println("Documentdirectory'"
+indexDir.getAbsolutePath()
+"'doesnotexistorisnotreadable,pleasecheckthepath");
System.exit(1);
}
try{
//创建索引库IndexWriter
if(writer==null){
initialIndexWriter(indexDir);
}
index(writer,order);
}catch(IOExceptione){
e.printStackTrace();
}
}
publicvoidClose()
{
if(null!=writer)
{
writer.close();
}
}
privatevoidinitialIndexWriter(FileindexDir)throwsIOException{
DirectoryreturnIndexDir=FSDirectory.open(indexDir);
IndexWriterConfigiwc=newIndexWriterConfig(Version.LUCENE_48,newStandardAnalyzer(Version.LUCENE_48));
writer=newIndexWriter(returnIndexDir,iwc);
}
privatevoidindex(IndexWriterwriter,OrderModelorder)throwsIOException{
//创建文档Document
Documentdoc=newDocument();
FieldorderidField=newStringField("orderid",order.orderid,Field.Store.YES);
doc.add(orderidField);
...
//向索引库中写入文档内容
writer.addDocument(doc);
}
}
}
七、组合条件查询
由java类实现查询，因为PHP可以直接调用java程序，所以PHP只需要把java类返回的结果显示出来就可以。
intpageIndex=1;
intpageSize=1;
intstart=(pageIndex-1)*pageSize;
Stringquery_fields[]=newString[]{"filename","content"};//对哪几个字段进行查询检索
Filefile_index_dir=newFile(indexDir);
try{
Directorydirectory=newSimpleFSDirectory(file_index_dir);
IndexReaderindexReader=DirectoryReader.open(directory);
//创建搜索类
IndexSearcherindexSearcher=newIndexSearcher(indexReader);
TermQueryquery1=newTermQuery(newTerm("orderid",orderid));
TermQueryquery2=newTermQuery(newTerm("contact",contact));
BooleanQueryquery=newBooleanQuery();
query.add(query1,BooleanClause.Occur.MUST);
query.add(query2,BooleanClause.Occur.MUST);
intmax_result_size=start+pageSize;
TopScoreDocCollectortopDocs=TopScoreDocCollector.create(max_result_size,false);
indexSearcher.search(query,topDocs);
introwCount=topDocs.getTotalHits();//满足条件的总记录数
intpages=(rowCount-1)/pageSize+1;//计算总页数
TopDocstds=topDocs.topDocs(start,pageSize);
ScoreDoc[]scoreDoc=tds.scoreDocs;
for(inti=0;i<scoreDoc.length;i++){
//内部编号
intdoc_id=scoreDoc[i].doc;
//根据文档id找到文档
Documentmydoc=indexSearcher.doc(doc_id);
//读取搜索结果
mydoc.get("orderid");
mydoc.get("contact");
mydoc.get("addr");
...
}
}catch(Exceptione){
e.printStackTrace();
}
有问题可以加我QQ沟通：10980327
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。