Storm系列(十九)普通事务ITransactionalSpout及示例

时间：2015-10-26 18:40:32 阅读：215 评论：0 收藏：0 [点我收藏+]

标签：

普通事务API详解

            1            public interface ITransactionalSpout<T> extends IComponent {       
         2                public interface Coordinator<X> {       
         3                    // 事务初始化       
         4                    X initializeTransaction(BigInteger txid, X prevMetadata);       
         5                    // 启动事务，返回true表示开始       
         6                    boolean isReady();              
         7                    // 结束时调用主要用于释放资源       
         8                    void close();       
         9                }       
         10                       
         11                public interface Emitter<X> {       
         12                    // 发射batch中的tuple到下一级Bolt       
         13                    void emitBatch(TransactionAttempt tx, X coordinatorMeta, BatchOutputCollector collector);       
         14                    // 根据事务ID进行状态数据的清理       
         15                    void cleanupBefore(BigInteger txid);       
         16                    // 结束时调用主要用于释放资源          
         17                    void close();       
         18                }       
         19                           
         20                Coordinator<T> getCoordinator(Map conf, TopologyContext context);       
         21                       
         22                Emitter<T> getEmitter(Map conf, TopologyContext context);       
         23            }       
 

示例

入口类

            1            public class TestMain {       
         2                    
         3                public static void main(String[] args) {       
         4                    
         5                    TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder(       
         6                            "ttbId", "spoutid", new TestTransaction(), 1);       
         7                    builder.setBolt("bolt1", new TestTransBolt1(), 3).shuffleGrouping(       
         8                            "spoutid");       
         9                    builder.setBolt("committer", new TestTransBolt2(), 1).shuffleGrouping(       
         10                            "bolt1");       
         11                    
         12                    Config conf = new Config();       
         13                    conf.setDebug(false);       
         14                    
         15                    if (args.length > 0) {       
         16                        try {       
         17                            StormSubmitter.submitTopology(args[0], conf,       
         18                                    builder.buildTopology());       
         19                        } catch (AlreadyAliveException e) {       
         20                            e.printStackTrace();       
         21                        } catch (InvalidTopologyException e) {       
         22                            e.printStackTrace();       
         23                        }       
         24                    } else {       
         25                        LocalCluster localCluster = new LocalCluster();       
         26                        localCluster.submitTopology("mytopology", conf,       
         27                                builder.buildTopology());       
         28                    }       
         29                }       
         30                    
         31            }       
 

普通事务spout

            1            public class TestTransaction implements ITransactionalSpout<TestMetaDate> {       
         2                    
         3                private static final long serialVersionUID = 1L;       
         4                private Map<Long, String> DATA_BASE = null;       
         5                    
         6                public TestTransaction(){       
         7                    DATA_BASE = new HashMap<Long, String>();       
         8                           
         9                    for (long i=0;i<50;i++){       
         10                        DATA_BASE.put(i, "TestTransaction:"+i);       
         11                    }       
         12                           
         13                    System.out.println("TestTransaction start");       
         14                }       
         15                       
         16                @Override       
         17                public void declareOutputFields(OutputFieldsDeclarer declarer) {       
         18                    declarer.declare(new Fields("tx","count"));       
         19                }       
         20                    
         21                @Override       
         22                public Map<String, Object> getComponentConfiguration() {       
         23                    return null;       
         24                }       
         25                    
         26                @Override       
         27                public backtype.storm.transactional.ITransactionalSpout.Coordinator<TestMetaDate> getCoordinator(       
         28                        Map conf, TopologyContext context) {       
         29                    System.out.println("TestTransaction getCoordinator start");       
         30                    return new TestCoordinator();       
         31                }       
         32                    
         33                @Override       
         34                public backtype.storm.transactional.ITransactionalSpout.Emitter<TestMetaDate> getEmitter(       
         35                        Map conf, TopologyContext context) {       
         36                    System.out.println("TestTransaction getEmitter start");       
         37                    return new TestEmitter(DATA_BASE);       
         38                }       
         39            }       
 

元数据实现类(存储到zookeeper中)

            1            public class TestMetaDate implements Serializable {       
         2                    
         3                private static final long serialVersionUID = 1L;       
         4                       
         5                private long _index;       
         6                private long _size;       
         7                       
         8                public long get_index() {       
         9                    return _index;       
         10                }       
         11                public void set_index(long _index) {       
         12                    this._index = _index;       
         13                }       
         14                       
         15                public long get_size() {       
         16                    return _size;       
         17                }       
         18                public void set_size(long _size) {       
         19                    this._size = _size;       
         20                }       
         21                       
         22                @Override       
         23                public String toString() {       
         24                    return "[_index=" + _index + ", _size=" + _size + "]";       
         25                }       
         26            }       
 

元数据协调处理类

            1            public class TestCoordinator implements ITransactionalSpout.Coordinator<TestMetaDate>{       
         2                    
         3                public TestCoordinator(){       
         4                    System.out.println("TestCoordinator start");       
         5                }       
         6                       
         7                @Override       
         8                public TestMetaDate initializeTransaction(BigInteger txid,       
         9                        TestMetaDate prevMetadata) {       
         10                    long index = 0L;       
         11                    if (null == prevMetadata){       
         12                        index = 0L;       
         13                    }       
         14                    else {       
         15                        index = prevMetadata.get_index()+prevMetadata.get_size();       
         16                    }       
         17                    TestMetaDate metaDate = new TestMetaDate();       
         18                    metaDate.set_index(index);       
         19                    metaDate.set_size(10);       
         20                    System.out.println("开始事务："+metaDate.toString());       
         21                    return metaDate;       
         22                }       
         23                    
         24                @Override       
         25                public boolean isReady() {       
         26                    Utils.sleep(1000);       
         27                    return true;       
         28                }       
         29                    
         30                @Override       
         31                public void close() {       
         32                }       
         33                       
         34            }       
 

Batch中的tuple发送处理类

            1            public class TestEmitter implements ITransactionalSpout.Emitter<TestMetaDate> {       
         2                    
         3                private Map<Long, String> _dbMap = null;       
         4                    
         5                public TestEmitter(Map<Long, String> dbMap) {       
         6                    System.err.println("start TestEmitter");       
         7                    this._dbMap = dbMap;       
         8                }       
         9                    
         10                @Override       
         11                public void emitBatch(TransactionAttempt tx, TestMetaDate coordinatorMeta,       
         12                        BatchOutputCollector collector) {       
         13                    long index = coordinatorMeta.get_index();       
         14                    long size = index + coordinatorMeta.get_size();       
         15                    System.err.println("TestEmitter emitBatch size:" + size       
         16                            + ",_dbMap size:" + _dbMap.size());       
         17                    if (size > _dbMap.size()) {       
         18                        return;       
         19                    }       
         20                    for (; index < size; index++) {       
         21                        if (null == _dbMap.get(index)) {       
         22                            System.out.println("TestEmitter continue");       
         23                            continue;       
         24                        }       
         25                        System.err.println("TestEmitter emitBatch index:"+index);       
         26                        collector.emit(new Values(tx, _dbMap.get(index)));       
         27                    }       
         28                }       
         29                    
         30                @Override       
         31                public void cleanupBefore(BigInteger txid) {       
         32                }       
         33                    
         34                @Override       
         35                public void close() {       
         36                }       
         37                    
         38            }       
 

数据单元统计实现类

            1            public class TestTransBolt1 extends BaseTransactionalBolt {       
         2                    
         3                private static final long serialVersionUID = 1L;       
         4                private BatchOutputCollector _outputCollector;       
         5                private TransactionAttempt _tx;       
         6                private int count = 0;       
         7                private TopologyContext _context;       
         8                    
         9                public TestTransBolt1() {       
         10                    System.out.println("start TestTransBolt1 ");       
         11                }       
         12                    
         13                @Override       
         14                public void prepare(@SuppressWarnings("rawtypes") Map conf,       
         15                        TopologyContext context, BatchOutputCollector collector,       
         16                        TransactionAttempt id) {       
         17                    this._context = context;       
         18                    this._outputCollector = collector;       
         19                    System.out.println("1 prepare TestTransBolt1 TransactionId:"       
         20                            + id.getTransactionId() + ",AttemptId:" + id.getAttemptId());       
         21                    
         22                }       
         23                    
         24                @Override       
         25                public void execute(Tuple tuple) {       
         26                    _tx = (TransactionAttempt) tuple.getValueByField("tx");       
         27                    String content = tuple.getStringByField("count");       
         28                    System.out.println("1 TaskId:"+_context.getThisTaskId()+",TestTransBolt1 TransactionAttempt "       
         29                            + _tx.getTransactionId() + "  attemptid" + _tx.getAttemptId());       
         30                    if (null != content && !content.isEmpty()) {       
         31                        count++;       
         32                    }       
         33                }       
         34                    
         35                @Override       
         36                public void finishBatch() {       
         37                    System.out.println("1 TaskId:"+_context.getThisTaskId()+",finishBatch count:"+count);       
         38                    _outputCollector.emit(new Values(_tx, count));       
         39                }       
         40                    
         41                @Override       
         42                public void declareOutputFields(OutputFieldsDeclarer declarer) {       
         43                    declarer.declare(new Fields("tx", "count"));       
         44                }       
         45                    
         46            }       
 

数据汇总统计实现类

            1            public class TestTransBolt2 extends BaseTransactionalBolt implements ICommitter {       
         2                    
         3                private static final long serialVersionUID = 1L;       
         4                private int sum = 0;       
         5                private TransactionAttempt _tx;       
         6                private static int _result = 0;       
         7                private static BigInteger _curtxid=null;       
         8                    
         9                public TestTransBolt2() {       
         10                    System.out.println("TestTransBolt2 start!");       
         11                }       
         12                    
         13                @Override       
         14                public void prepare(@SuppressWarnings("rawtypes") Map conf,       
         15                        TopologyContext context, BatchOutputCollector collector,       
         16                        TransactionAttempt id) {       
         17                    
         18                    this._tx = id;       
         19                    System.out.println("TestTransBolt2 prepare TransactionId:" + id);       
         20                }       
         21                    
         22                @Override       
         23                public void execute(Tuple tuple) {       
         24                    _tx = (TransactionAttempt) tuple.getValueByField("tx");       
         25                    sum += tuple.getIntegerByField("count");       
         26                    
         27                    System.out.println("TestTransBolt2 execute TransactionAttempt:" + _tx);       
         28                }       
         29                    
         30                @Override       
         31                public void finishBatch() {       
         32                    System.out.println("finishBatch _curtxid:" + _curtxid       
         33                            + ",getTransactionId:" + _tx.getTransactionId());       
         34                    if (null == _curtxid || !_curtxid.equals(_tx.getTransactionId())) {       
         35                    
         36                        System.out.println("****** 1 _curtxid:" + _curtxid       
         37                                + ",_tx.getTransactionId():" + _tx.getTransactionId());       
         38                    
         39                        if (null == _curtxid) {       
         40                            _result = sum;       
         41                        } else {       
         42                            _result += sum;       
         43                        }       
         44                        _curtxid = _tx.getTransactionId();       
         45                        System.out.println("****** 2 _curtxid:" + _curtxid       
         46                                + ",_tx.getTransactionId():" + _tx.getTransactionId());       
         47                    }       
         48                    
         49                    System.out.println("total==========================:" + _result);       
         50                }       
         51                    
         52                @Override       
         53                public void declareOutputFields(OutputFieldsDeclarer declarer) {       
         54                    
         55                }       
         56            }       
 

结果：

TestTransBolt2 prepare TransactionId:5:3709460125605136723
TestTransBolt2 execute TransactionAttempt:5:3709460125605136723
TestTransBolt2 execute TransactionAttempt:5:3709460125605136723
TestTransBolt2 execute TransactionAttempt:5:3709460125605136723
finishBatch _curtxid:4,getTransactionId:5
****** 1 _curtxid:4,_tx.getTransactionId():5
****** 2 _curtxid:5,_tx.getTransactionId():5
total==========================:50
开始事务：[_index=50, _size=10]
TestEmitter emitBatch size:60,_dbMap size:50

Storm系列(十九)普通事务ITransactionalSpout及示例

标签：

原文地址：http://www.cnblogs.com/jianyuan/p/4911742.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行

1	public interface ITransactionalSpout<T> extends IComponent {
2	public interface Coordinator<X> {
3	// 事务初始化
4	X initializeTransaction(BigInteger txid, X prevMetadata);
5	// 启动事务，返回true表示开始
6	boolean isReady();
7	// 结束时调用主要用于释放资源
8	void close();
9	}
10
11	public interface Emitter<X> {
12	// 发射batch中的tuple到下一级Bolt
13	void emitBatch(TransactionAttempt tx, X coordinatorMeta, BatchOutputCollector collector);
14	// 根据事务ID进行状态数据的清理
15	void cleanupBefore(BigInteger txid);
16	// 结束时调用主要用于释放资源
17	void close();
18	}
19
20	Coordinator<T> getCoordinator(Map conf, TopologyContext context);
21
22	Emitter<T> getEmitter(Map conf, TopologyContext context);
23	}

1	public class TestMain {
2
3	public static void main(String[] args) {
4
5	TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder(
6	"ttbId", "spoutid", new TestTransaction(), 1);
7	builder.setBolt("bolt1", new TestTransBolt1(), 3).shuffleGrouping(
8	"spoutid");
9	builder.setBolt("committer", new TestTransBolt2(), 1).shuffleGrouping(
10	"bolt1");
11
12	Config conf = new Config();
13	conf.setDebug(false);
14
15	if (args.length > 0) {
16	try {
17	StormSubmitter.submitTopology(args[0], conf,
18	builder.buildTopology());
19	} catch (AlreadyAliveException e) {
20	e.printStackTrace();
21	} catch (InvalidTopologyException e) {
22	e.printStackTrace();
23	}
24	} else {
25	LocalCluster localCluster = new LocalCluster();
26	localCluster.submitTopology("mytopology", conf,
27	builder.buildTopology());
28	}
29	}
30
31	}

1	public class TestTransaction implements ITransactionalSpout<TestMetaDate> {
2
3	private static final long serialVersionUID = 1L;
4	private Map<Long, String> DATA_BASE = null;
5
6	public TestTransaction(){
7	DATA_BASE = new HashMap<Long, String>();
8
9	for (long i=0;i<50;i++){
10	DATA_BASE.put(i, "TestTransaction:"+i);
11	}
12
13	System.out.println("TestTransaction start");
14	}
15
16	@Override
17	public void declareOutputFields(OutputFieldsDeclarer declarer) {
18	declarer.declare(new Fields("tx","count"));
19	}
20
21	@Override
22	public Map<String, Object> getComponentConfiguration() {
23	return null;
24	}
25
26	@Override
27	public backtype.storm.transactional.ITransactionalSpout.Coordinator<TestMetaDate> getCoordinator(
28	Map conf, TopologyContext context) {
29	System.out.println("TestTransaction getCoordinator start");
30	return new TestCoordinator();
31	}
32
33	@Override
34	public backtype.storm.transactional.ITransactionalSpout.Emitter<TestMetaDate> getEmitter(
35	Map conf, TopologyContext context) {
36	System.out.println("TestTransaction getEmitter start");
37	return new TestEmitter(DATA_BASE);
38	}
39	}

1	public class TestMetaDate implements Serializable {
2
3	private static final long serialVersionUID = 1L;
4
5	private long _index;
6	private long _size;
7
8	public long get_index() {
9	return _index;
10	}
11	public void set_index(long _index) {
12	this._index = _index;
13	}
14
15	public long get_size() {
16	return _size;
17	}
18	public void set_size(long _size) {
19	this._size = _size;
20	}
21
22	@Override
23	public String toString() {
24	return "[_index=" + _index + ", _size=" + _size + "]";
25	}
26	}

1	public class TestCoordinator implements ITransactionalSpout.Coordinator<TestMetaDate>{
2
3	public TestCoordinator(){
4	System.out.println("TestCoordinator start");
5	}
6
7	@Override
8	public TestMetaDate initializeTransaction(BigInteger txid,
9	TestMetaDate prevMetadata) {
10	long index = 0L;
11	if (null == prevMetadata){
12	index = 0L;
13	}
14	else {
15	index = prevMetadata.get_index()+prevMetadata.get_size();
16	}
17	TestMetaDate metaDate = new TestMetaDate();
18	metaDate.set_index(index);
19	metaDate.set_size(10);
20	System.out.println("开始事务："+metaDate.toString());
21	return metaDate;
22	}
23
24	@Override
25	public boolean isReady() {
26	Utils.sleep(1000);
27	return true;
28	}
29
30	@Override
31	public void close() {
32	}
33
34	}

1	public class TestEmitter implements ITransactionalSpout.Emitter<TestMetaDate> {
2
3	private Map<Long, String> _dbMap = null;
4
5	public TestEmitter(Map<Long, String> dbMap) {
6	System.err.println("start TestEmitter");
7	this._dbMap = dbMap;
8	}
9
10	@Override
11	public void emitBatch(TransactionAttempt tx, TestMetaDate coordinatorMeta,
12	BatchOutputCollector collector) {
13	long index = coordinatorMeta.get_index();
14	long size = index + coordinatorMeta.get_size();
15	System.err.println("TestEmitter emitBatch size:" + size
16	+ ",_dbMap size:" + _dbMap.size());
17	if (size > _dbMap.size()) {
18	return;
19	}
20	for (; index < size; index++) {
21	if (null == _dbMap.get(index)) {
22	System.out.println("TestEmitter continue");
23	continue;
24	}
25	System.err.println("TestEmitter emitBatch index:"+index);
26	collector.emit(new Values(tx, _dbMap.get(index)));
27	}
28	}
29
30	@Override
31	public void cleanupBefore(BigInteger txid) {
32	}
33
34	@Override
35	public void close() {
36	}
37
38	}

1	public class TestTransBolt1 extends BaseTransactionalBolt {
2
3	private static final long serialVersionUID = 1L;
4	private BatchOutputCollector _outputCollector;
5	private TransactionAttempt _tx;
6	private int count = 0;
7	private TopologyContext _context;
8
9	public TestTransBolt1() {
10	System.out.println("start TestTransBolt1 ");
11	}
12
13	@Override
14	public void prepare(@SuppressWarnings("rawtypes") Map conf,
15	TopologyContext context, BatchOutputCollector collector,
16	TransactionAttempt id) {
17	this._context = context;
18	this._outputCollector = collector;
19	System.out.println("1 prepare TestTransBolt1 TransactionId:"
20	+ id.getTransactionId() + ",AttemptId:" + id.getAttemptId());
21
22	}
23
24	@Override
25	public void execute(Tuple tuple) {
26	_tx = (TransactionAttempt) tuple.getValueByField("tx");
27	String content = tuple.getStringByField("count");
28	System.out.println("1 TaskId:"+_context.getThisTaskId()+",TestTransBolt1 TransactionAttempt "
29	+ _tx.getTransactionId() + " attemptid" + _tx.getAttemptId());
30	if (null != content && !content.isEmpty()) {
31	count++;
32	}
33	}
34
35	@Override
36	public void finishBatch() {
37	System.out.println("1 TaskId:"+_context.getThisTaskId()+",finishBatch count:"+count);
38	_outputCollector.emit(new Values(_tx, count));
39	}
40
41	@Override
42	public void declareOutputFields(OutputFieldsDeclarer declarer) {
43	declarer.declare(new Fields("tx", "count"));
44	}
45
46	}

1	public class TestTransBolt2 extends BaseTransactionalBolt implements ICommitter {
2
3	private static final long serialVersionUID = 1L;
4	private int sum = 0;
5	private TransactionAttempt _tx;
6	private static int _result = 0;
7	private static BigInteger _curtxid=null;
8
9	public TestTransBolt2() {
10	System.out.println("TestTransBolt2 start!");
11	}
12
13	@Override
14	public void prepare(@SuppressWarnings("rawtypes") Map conf,
15	TopologyContext context, BatchOutputCollector collector,
16	TransactionAttempt id) {
17
18	this._tx = id;
19	System.out.println("TestTransBolt2 prepare TransactionId:" + id);
20	}
21
22	@Override
23	public void execute(Tuple tuple) {
24	_tx = (TransactionAttempt) tuple.getValueByField("tx");
25	sum += tuple.getIntegerByField("count");
26
27	System.out.println("TestTransBolt2 execute TransactionAttempt:" + _tx);
28	}
29
30	@Override
31	public void finishBatch() {
32	System.out.println("finishBatch _curtxid:" + _curtxid
33	+ ",getTransactionId:" + _tx.getTransactionId());
34	if (null == _curtxid \|\| !_curtxid.equals(_tx.getTransactionId())) {
35
36	System.out.println("**** 1 _curtxid:**" + _curtxid
37	+ ",_tx.getTransactionId():" + _tx.getTransactionId());
38
39	if (null == _curtxid) {
40	_result = sum;
41	} else {
42	_result += sum;
43	}
44	_curtxid = _tx.getTransactionId();
45	System.out.println("**** 2 _curtxid:**" + _curtxid
46	+ ",_tx.getTransactionId():" + _tx.getTransactionId());
47	}
48
49	System.out.println("total==========================:" + _result);
50	}
51
52	@Override
53	public void declareOutputFields(OutputFieldsDeclarer declarer) {
54
55	}
56	}