Storm项目：流数据监控（下）项目：流数据监控（下）
1 文档说明
该文档为storm模拟项目系列文档之一，是MetaQ与storm接口的说明文档，主要介绍了如何集成MetaQ到项目代码中。
软件（阿里），其对应的许多技术文档还是比较容易看的，并且Github提供了许多的应用实例，所以使用MetaQ作为Storm的
消息源之一是必须掌握的。
该模拟项目中，使用MetaQ作为storm的消息源，MetaqSpout从指定的zkconnect及topic中读取数据，发布到节点中。此外，
还写了MetaQ与storm的生产者接口，即MetaqBolt指定Topic将数据写入Metaq中供其他业务系统继续使用。
好了文档说明就这些了，代码马上会随文档更新（通常情况下，代码调试成功了我才会写文档的）。
2 MetaQ与Storm接口
2.1 MetaqSpout
2.1.1 接口说明
该接口参考自Github，作了部分修改。项目设计中，使用storm.xml.MetaqSpoutXml读取MetaqSpout对应的配置文件
MetaqSpout.xml
配置文件中，指明zkconnect的地址及端口号、metaq的root目录、对应的消费topic及其消费组（这个很重要）。
读取配置之后，将配置传递到spout的open部分进行初始化工作，主要是进行消费者参数设定（包括zkconnect、root目录、
Topic及Group设置）等。
在nextTuple方法中，进行消息（message）拉取（poll），每次拉取一条记录，发布到下一个拓扑节点中。
2.1.2 上代码
贴部分主要代码（详细参考代码包）：
//构造函数，传递xml地址
public MetaqSpout(String MetaqSpoutXml) {
super();
this.metaqspoutxml = MetaqSpoutXml;
}
//实例化参数配置类
private ZKConfig zkConfig = new ZKConfig();
private MetaClientConfig metaClientConfig = new MetaClientConfig();
private final Scheme scheme = new StringScheme();
//初始化调用
public void open(final Map conf, final TopologyContext context,
final SpoutOutputCollector collector) {
//从xml中获取参数
new MetaqSpoutXml(this.metaqspoutxml).read();
this.zkConfig.zkConnect = MetaqSpoutXml.zkConnect;//"192.168.2.240:2181";
this.zkConfig.zkRoot = MetaqSpoutXml.zkRoot;//"/meta";
String topic = MetaqSpoutXml.topic;
String group = MetaqSpoutXml.group;
this.metaClientConfig.setZkConfig(this.zkConfig);
this.consumerConfig = new ConsumerConfig(group);
//final String topic = (String) conf.get(TOPIC);
if (topic == null) {
throw new IllegalArgumentException(TOPIC + " is null");
}
Integer maxSize = (Integer) conf.get(FETCH_MAX_SIZE);
剩余14页未读，继续阅读
资源评论
weixin_38535812
- 粉丝: 5
最新资源
资源上传下载、课程学习等过程中有任何疑问或建议，欢迎提出宝贵意见哦~我们会及时处理！
点击此处反馈
