(一）kafka-jstorm集群实时日志分析 之 ---------kafka实时日志处理-----续

上文(一）kafka-jstorm集群实时日志分析 之 ---------kafka实时日志处理-　讲述了日志如何写入kafka集群，现在看一下效果．

zookeeper集群，kafka集群启动后，我们用kafka-web-console (https://github.com/claudemamo/kafka-web-console) 观察一下集群运行状态．

配置好kafka-web-console，观察一下zookeeper集群，三个zookeeper的状态，如下图所示，Status状态为已连接．

观察一下kafka集群情况，如下图，二个kafka都已注册在zookeeper集群．

　　观察一下kafak　topic　，如下图：下图是还没启动消息日志，这是上次遗留的状态．

　　　好，我们启动日志消息生产者，KafkaAppenderTest测试入口，先运行junit　ｔｅｓｔpublic void test_log_producer()　　方法，

然后运行public void test_comsumer()方法．

控制台信息如下：日志消息producer,已向kafka集群生产日志信息．

<span style="font-size:18px;">while(true){
			LOG.info("test_log_producer : "  + RandomStringUtils.random(3, "hello doctro,how are you,and you"));
		}</span>

　看一下kafka消息消费者接收到的日志信息：

<span style="font-size:18px;">			while(it.hasNext())
			System.out.println(new String("test_comsumer: " + new String(it.next().message())));</span>

　证明日志消息写入了kafak集群，消费者也可以从kafka集群消费到这些实时日志信息．

　下次，把整合jstorm的东东贴上．