标签:ping log 逻辑 概念 .com lse orm 两种 ima
Strom框架基本概念就不提了,这里主要讲的是Stream
自定义ID的消息流。默认spout、bolt都需实现接口方法declareOutputFields
,代码如下:
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("body"));
}
这种情况下发的消息会被所有定义的bolts接收。我们如果需要根据得到的消息类型来选择不同的bolt,就需要用到Stream Grouping。
OutputFieldsDeclarer
来定义发射多条消息流stream以下定义了两种stream消息流:email邮件、sms短信
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declareStream("email", new Fields("body"));
outputFieldsDeclarer.declareStream("sms", new Fields("body"));
}
@Override
public void execute(Tuple tuple) {
String streamType;
String value = tuple.getStringByField("body");
# 逻辑判断stub code
if (value.startsWith("email:")) {
streamType = "email";
} else {
streamType = "sms";
}
outputCollector.emit(streamType, new Values(value));
}
FilterBolt通过对消息进行加工处理,下发给bolts时会指定不同的stream,EmailNotifyBolt只接收email
类型的stream消息,SmsNotifyBolt只接收sms
类型的stream消息。
TopologyBuilder topologyBuilder = new TopologyBuilder();
topologyBuilder.setSpout("RabbitmqSpout", new RabbitmqSpout());
topologyBuilder.setBolt("FilterBolt", new FilterBolt()).shuffleGrouping("RabbitmqSpout");
topologyBuilder.setBolt("EmailNotifyBolt", new EmailNotifyBolt()).localOrShuffleGrouping("FilterBolt", "email");
topologyBuilder.setBolt("SmsNotifyBolt", new SmsNotifyBolt()).localOrShuffleGrouping("FilterBolt", "sms");
Storm框架:如何根据业务条件选择不同的bolt进行下发消息
标签:ping log 逻辑 概念 .com lse orm 两种 ima
原文地址:https://www.cnblogs.com/gouyg/p/java_storm_stream.html