getting start with storm 翻译第六章 part-4

转载请注明出处：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 通信，并且都以相同的标准，确保引用相同产品的请求在相同的服务器上结束的情况下才做的到。