码迷,mamicode.com
首页 > 其他好文 > 详细

Storm框架:如何根据业务条件选择不同的bolt进行下发消息

时间:2018-11-02 01:55:46      阅读:205      评论:0      收藏:0      [点我收藏+]

标签: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"));
}
  • 然后我们通过对消息内容进行分析判断来决定发射指定的stream类型
@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));
}
  • topology设置bolt的消息源时通过localOrShuffleGrouping来设置只接收指定stream的消息

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

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!