14.3 Strom测试代码

忘是亡心i 2022-05-14 11:24 160阅读 0赞

![70][]

StromTest ：

package com.yuntian.test;

import backtype.storm.Config;

import backtype.storm.LocalCluster;

import backtype.storm.StormSubmitter;

import backtype.storm.generated.AlreadyAliveException;

import backtype.storm.generated.InvalidTopologyException;

import backtype.storm.topology.TopologyBuilder;

public class StromTest \{

public static void main(String\[\] args) throws AlreadyAliveException, InvalidTopologyException\{

TopologyBuilder bulider = new TopologyBuilder();

bulider.setSpout("myspout", new MySpout());

bulider.setBolt("mybolt", new MyBolt()).shuffleGrouping("myspout");

Config conf = new Config();

//当有参数的时候是集群执行

if(args !=null && args.length > 0)\{

conf.setNumWorkers(3);

StormSubmitter.submitTopology("StromTest", conf, bulider.createTopology());

\}else\{

conf.setDebug(true);

conf.setMaxSpoutPending(10);//多少个线程来启动

LocalCluster locla = new LocalCluster();

locla.submitTopology("StromTest", conf, bulider.createTopology());

--------------------

MySpout：

package com.yuntian.test;

import java.util.Map;

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichSpout;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

public class MySpout implements IRichSpout\{

SpoutOutputCollector collector = null;

@Override

public void ack(Object arg0) \{

@Override

public void activate() \{

@Override

public void close() \{

@Override

public void deactivate() \{

@Override

public void fail(Object arg0) \{

@Override

public void nextTuple() \{

try \{

Thread.sleep(1000);

\} catch (InterruptedException e) \{

e.printStackTrace();

collector.emit(new Values("111","222","333"));

@Override

public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) \{

this.collector=collector;

@Override

public void declareOutputFields(OutputFieldsDeclarer arg0) \{

arg0.declare(new Fields("aaa","bbb","ccc"));

@Override

public Map<String, Object> getComponentConfiguration() \{

return null;

--------------------

MyBolt：

package com.yuntian.test;

import java.util.Map;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichBolt;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Tuple;

public class MyBolt implements IRichBolt\{

@Override

public void cleanup() \{

@Override

public void execute(Tuple arg0) \{

System.out.println(arg0.getString(0));

System.out.println(arg0.getString(1));

System.out.println(arg0.getString(2));

@Override

public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) \{

@Override

public void declareOutputFields(OutputFieldsDeclarer arg0) \{

@Override

public Map<String, Object> getComponentConfiguration() \{

return null;

--------------------

本机连接storm集群在eclipse上测试运行：

![70 1][]

[70]: /images/20220514/50268bc4fe2c42e1be611a82624bd912.png
[70 1]: /images/20220514/6463497796d54d6a9a636bd4299d7bc8.png

14.3 Strom测试代码

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Strom-实时流计算框架

相关 Strom-实时流计算框架

相关 [leetcode]143. Reorder List -- JavaScript 代码

相关 Strom官网信息

相关 strom环境搭建

相关 14.3 Strom测试代码

相关 strom-kafka简单整合

相关 143. Reorder List

相关 strom部署问题

相关 Strom优化指南