版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/Simon_09010817/article/details/81363822
一、概述
全局分组方式将所有的 tuples 路由到唯一一个 task 上。Storm选取最小的 task ID 来选取接收数据的 task。注意,当使用全局分组时,设置 bolt 的 task 并发度是没有意义的,因为所有 tuple 都转发到同一个 task 上了。使用全局分组的时候需要注意,因为所有的tuple都转发到一个JVM实例上,可能会引起Storm集群中某个JVM或者服务器出现性能瓶颈或崩溃。
二、代码
- Spout
package com.test.csdn.globalgrouping;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import java.util.Map;
import java.util.Random;
/**
*
* @author Simon
* @date 2018/8/1
*/
public class GlobalGroupingSpout extends BaseRichSpout {
private SpoutOutputCollector collector;
private String[] str = {"xiaomi","huawei","apple","oppo","vivo","lenovo","LG",
"samsung","htc","honor","nokia","smartisan","Sony","BlackBerry","sharp"};
@Override
public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
this.collector=spoutOutputCollector;
}
@Override
public void nextTuple() {
try {
Thread.sleep(1000);
int i = new Random().nextInt(10);
String string = str[i];
collector.emit(new Values(string));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("string"));
}
}
2.Bolt
package com.test.csdn.globalgrouping;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
/**
* Created by Simon on 2018/8/1.
*/
public class GlobalGroupingBolt extends BaseBasicBolt {
@Override
public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
System.out.println(Thread.currentThread().getName()+"___"+tuple.getValue(0));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
}
}
3.Topo
package com.test.csdn.globalgrouping;
import com.test.csdn.fieldsgrouping.FiledGroupingBolt;
import com.test.csdn.fieldsgrouping.FiledGroupingSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
/**
* Created by Simon on 2018/8/1.
*/
public class GlobalGroupingTopo {
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new GlobalGroupingSpout()).setNumTasks(3);
builder.setBolt("bolt", new GlobalGroupingBolt()).setNumTasks(5).globalGrouping("spout");
Config conf = new Config();
conf.setDebug(false);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("toplogy", conf, builder.createTopology());
Utils.sleep(Long.MAX_VALUE);
cluster.shutdown();
}
}
三、运行输出
运行在同一个taskId里面。