转载请注明出处：http://blog.csdn.net/lonelytrooper/article/details/9982967
Redis服务器
Redis是一套高级的用于持久化的内存KeyValue存储系统(见http://redis.io)。使用它来存储下述信息：
﹒产品信息，用于服务网站。
﹒用户导航队列，用于供给Storm Topology。
﹒Storm Topology中间数据，用于Topology从失败中恢复。
﹒Storm Topology结果，用于存储预期的结果。
生产信息
Redis服务器存储产品，它使用产品ID作为键，包含所有产品信息的JSON对象作为值。
用户导航队列
用户导航队列被存储在一个命名为导航的Redis列表中并且被组织成一个先进先出(FIFO)的队列。每次用户访问一个产品页，服务器添加一个项到列表左端，以此来表明哪个用户访问了哪个商品。Storm集群不时地从列表的右端删除元素来处理信息。
中间数据
集群需要分别存储每个用户的历史记录。为达到这个目的，它在Redis服务器中保存了一个集合，该集合存储了每个用户浏览的所有产品及它们的分类。
结果
集群产生用户访问特定产品的有用数据并且将它们存储在命名为”procnt”的Redis Hash中：后边紧跟着产品ID。
测试Topology
为了测试topology，使用提供的LocalCluster和一个本地的Redis服务器(见图6-7)。你将在初始化时填充产品数据库并且在Redis服务器上模拟浏览日志的插入。我们的断言将通过读取topology输出到Redis服务器来执行。测试用Java和Groovy编写。
图6-7. 测试架构
测试初始化
初始化由三步组成：
启动LocalCluster并提交Topology. 初始化在AbstractAnalyticsTest中被执行，该类被所有的测试继承。一个叫做topologyStarted的静态标志被用来避免当多个AbstractAnalyticsTest子类初始化时AbstractAnalyticsTest本身被初始化不止一次的情况。
注意那里sleep的目的是允许LocalCluster在尝试从中恢复结果之前正确的启动。
public abstract class AbstractAnalyticsTest extendsAssert {
def jedis
static topologyStarted= false
static sync=new Object()
private voidreconnect(){
jedis =new Jedis(TopologyStarter.REDIS_HOST,TopologyStarter.REDIS_PORT)
}
@Before
public voidstartTopology(){
synchronized(sync){
reconnect()
if(!topologyStarted){
jedis.flushAll()
populateProducts()
TopologyStarter.testing= true
TopologyStarter.main(null)
topologyStarted =true
sleep 1000
}
}
}
...
public voidpopulateProducts(){
def testProducts = [
[id:0, title:"Dvdplayer with surround sound system",
category:"Players",price: 100],
[id:1, title:"FullHD Bluray and DVD player",
category:"Players",price:130],
[id:2, title:"Mediaplayer with USB 2.0 input",
category:"Players",price:70],
...
[id:21, title:"TVWall mount bracket 50-55 Inches",
category:"Mounts",price:80]
]
testProducts.each(){product ->
def val =
"{ \"title\":\"${product.title}\" , \"category\":\"${product.category}\","+
" \"price\": ${product.price},\"id\": ${product.id} }"
println val
jedis.set(product.id.toString(),val.toString())
}
}
...
}
在AbstractAnalyticsTest类中实现一个叫做navigate的方法。为了使不同的测试有一种来模拟用户导航页面行为的方式，该步在Redis服务器导航队列中插入导航项。
public abstract class AbstractAnalyticsTest extendsAssert {
...
public voidnavigate(user,product) {
String nav =
"{\"user\": \"${user}\",\"product\": \"${product}\", \"type\":\"PRODUCT
\"}".toString()
println "Pushingnavigation: ${nav}"
jedis.lpush('navigation',nav)
}
...
}
在AbstractAnalyticsTest中提供一个叫做getProductCategory的方法来从Redis服务器中读取特定的关系。不同的测试也需要对统计的结果进行断言来确保topology按预期的运行。
public abstract class AbstractAnalyticsTest extendsAssert {
...
public intgetProductCategoryStats(Stringproduct,Stringcateg) {
String count =jedis.hget("prodcnt:${product}",categ)
if(count== null|| "nil".equals(count))
return 0
return Integer.valueOf(count)
}
...
}
一个测试用例
在下边的小片段断中，你将模拟用户”1”的一些产品浏览记录，然后核实结果。注意在断言确定结果已经被存储到Redis之前你要等待两秒钟。(需要记住的是ProductCategoriesCounterBolt包含一个计数器的内存拷贝并且在后台将他们发送至Redis)。
package functional
class StatsTestextends AbstractAnalyticsTest{
@Test
public voidtestNoDuplication(){
navigate("1","0") // Players
navigate("1","1") // Players
navigate("1","2") // Players
navigate("1","3") // Cameras
Thread.sleep(2000)// Give two seconds for the system to process the data.
assertEquals 1,getProductCategoryStats("0","Cameras")
assertEquals 1,getProductCategoryStats("1","Cameras")
assertEquals 1,getProductCategoryStats("2","Cameras")
assertEquals 2,getProductCategoryStats("0","Players")
assertEquals 3,getProductCategoryStats("3","Players")
}
}
扩展性和可用性的说明
为适应本书一个单独章节的大小，该方案的架构被简化了。由于这个原因，你规避了一些对于这个方案的扩展和高可用性来说所必要的复杂性。该架构有两个主要的问题。
该架构中Redis服务器不仅仅是单点失败的而且是一个瓶颈。你只能获取Redis服务器所能处理的数据量。Redis层可以通过使用分片来扩展，并且它的可用性可以通过使用一个Master/Slave配置来改进，这需要topology和web应用资源都做出改变。
另一个弱点是以循环的方式添加机器时，web应用并没有成比例的扩展。这是因为当产品的统计改变的时候它需要被通知，并且是通知所有相应的浏览器。这个”通知浏览器”的桥接使用Socket.io实现，但它需要监听器和通知器被部署在同一台web服务器上。这只有在你共享GET /product/:id/stats 通信和 the POST /news 通信，并且都以相同的标准，确保引用相同产品的请求在相同的服务器上结束的情况下才做的到。
