一、实时业务指标分析
1.业务
业务：
订单系统---->MQ---->Kakfa--->Storm
数据：订单编号、订单时间、支付编号、支付时间、商品编号、商家名称、商品价格、优惠价格、支付金额
统计双十一当前的订单金额，订单数量，订单人数
订单金额（整个网站，各个业务线，各个品类，各个店铺，各个品牌，每个商品
架构
支付系统+kafka+storm/Jstorm集群+redis集群
1、支付系统发送mq到kafka集群中，编写storm程序消费kafka的数据并计算实时的订单数量、订单数量
2、将计算的实时结果保存在redis中
3、外部程序访问redis进群中的数据实时展示结果
流程
1、Spout获取外部数据源，数据源是订单的mq，mq有固定的格式，比如json串。
2、对订单mq进行解析，得到一个对象->JavaBean
订单编号、订单时间、支付编号、支付时间、商品编号、商家名称、商品价格、优惠价格、支付金额
3、对指标进行计数
//业务中一个订单包含多个商品，需要对每个商品进行指标计算
//创建订单和取消订单两种类型，在计算总数据的是考虑将取消订单的金额减掉
//订单中有拆单的逻辑，该如何计算
4、保存指标数据到Redis
2.整合flume思路
flume官网也是对应的kafka的sink配置与讲解：http://flume.apache.org/FlumeUserGuide.html#kafka-sink
（网上的一些例如brokelist等配置已经被标记为deprecated，请参照官网的最新属性配置）
将采集的日志文件保存到kafka中
（source） 输入：tail -F xxxx.log
（channel）存储：内存
（sink） 输出：kafka
config
al.source = s1
a1.channel = c1
al.sink = k1
source ==> exec tail -F xxxx.log
channel ==> RAM
sink ====> xxx.xxxx.xxxx.KafkaSink //该类必须存放lib目录
sink.topic = orderMq
sink.itcast = itcast
// 实例参照下文
二、整合flume
1.编写flume配置文件
a1.sources = s1
a1.channels = c1
a1.sinks = k1
a1.sources.s1.type=exec
a1.sources.s1.command=tail -F /home/hadoop/kafka.log
a1.sources.s1.channels=c1
a1.channels.c1.type=memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity=100
#设置Kafka接收器
a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink
#设置Kafka的broker地址和端口号
a1.sinks.k1.brokerList=mini1:9092
#设置Kafka的Topic
a1.sinks.k1.topic=topic_1
#设置序列化方式
a1.sinks.k1.serializer.class=kafka.serializer.StringEncoder
a1.sinks.k1.channel=c1
2.开启需要的zk以及kafka服务
[hadoop@mini1 conf]$ start-zk.sh
[hadoop@mini1 conf]$ start-kafka.sh
// 以上两个均以编写一键启动脚本，详情参考相关篇章随笔
启动flume：
bin/flume-ng agent -c conf -f conf/tail-kafka.conf -n a1 -Dflume.root.logger=INFO,console
启动消费者查看效果：
bin/kafka-console-consumer.sh --zookeeper mini1:2181 --from-beginning --topic topic_1
之前都是直接复制命令而不关心参数，通过官网查看可以发现查看完整参数的方法为：（网上找的基本都是这个常用参数的格式）
All of the command line tools have additional options; running the command with no arguments will display usage information documenting them in more detail.
这样就说明sink到了kafka了！
三、整合storm
1.maven依赖
storm整合kafka需要一个中间的依赖Jar
<!-- https://mvnrepository.com/artifact/org.apache.storm/storm-kafka -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>0.9.5</version>
</dependency>
这里指出，storm入门中使用的storm的包，应当替换为阿里的更加强大的JStorm：
<!-- https://mvnrepository.com/artifact/com.alibaba.jstorm/jstorm-core -->
<dependency>
<groupId>com.alibaba.jstorm</groupId>
<artifactId>jstorm-core</artifactId>
<version>2.1.1</version>
<!--<scope>provided</scope>-->
</dependency>
其他相关依赖如下：
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.36</version>
</dependency>
<!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.7.3</version>
</dependency>
这样，将数据提前导入redis以后，就可以写出一个demo版本的代码了：（注意MyBolt#exector()方法中需要手动调用collection.ack()进行应答）
package kafkaAndStorm; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.topology.TopologyBuilder; import storm.kafka.KafkaSpout; import storm.kafka.SpoutConfig; import storm.kafka.ZkHosts; public class KafkaAndStormTopologyMain { public static void main(String[] args) throws Exception{ TopologyBuilder topologyBuilder = new TopologyBuilder(); topologyBuilder.setSpout("kafkaSpout", new KafkaSpout(new SpoutConfig( new ZkHosts("zk01:2181,zk02:2181,zk03:2181"), "orderMq", "/myKafka", "kafkaSpout")),1); topologyBuilder.setBolt("mybolt1",new ParserOrderMqBolt(),1).shuffleGrouping("kafkaSpout"); Config config = new Config(); config.setNumWorkers(1); //3、提交任务 -----两种模式 本地模式和集群模式 if (args.length>0) { StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology()); }else { LocalCluster localCluster = new LocalCluster(); localCluster.submitTopology("storm2kafka", config, topologyBuilder.createTopology()); } } }