如何使用Apache Storm Tuple

使用所有字段作为值创建元组的另一种方法是只创建一个bean并将其传递给元组。

给定以下类别：

public class DataBean implements Serializable {    private static final long serialVersionUID = 1L;    // add more properties as necessary    int id;    String word;    public DataBean(int id, String word) {        setId(id);        setWord(word);    }    public int getId() {        return id;    }    public void setId(int id) {        this.id = id;    }    public String getWord() {        return word;    }    public void setWord(String word) {        this.word = word;    }}

一键创建并发出DataBean：

collector.emit(new Values(bean));

在目标螺栓中获取DataBean：

@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {    try {        DataBean bean = (DataBean)tuple.getValue(0);        // do your bolt processing with the bean    } catch (Exception e) {        LOG.error("WordCountBolt error", e);        collector.reportError(e);    }       }

设置拓扑时，请不要忘记使bean可序列化并注册：

Config stormConfig = new Config();stormConfig.registerSerialization(DataBean.class);// more stuffStormSubmitter.submitTopology("MyTopologyName", stormConfig, builder.createTopology());

免责声明：Beans可以很好地用于随机分组。如果需要执行

fieldsGrouping

collector.emit(new Values(word, bean));