storm事务

 1 package com.dxss.storm.transaction;
 2 
 3 import java.util.HashMap;
 4 import java.util.Map;
 5 import com.dsxx.storm.sql.utils.CreateData;
 6 import backtype.storm.task.TopologyContext;
 7 import backtype.storm.topology.OutputFieldsDeclarer;
 8 import backtype.storm.transactional.ITransactionalSpout;
 9 import backtype.storm.tuple.Fields;
10 
11 /**
12  * 
13  * @ClassName: TransSpout
14  * @Description: Storm的事务性操作,ITransactionalSpout 获取数据并向拓扑分发批次,
15  *                   泛型MyMata(元数据: 包含当前事务可以从哪个point进行重发数据，放在zookeeper中的，spout可以通过kryo从
16  *                  zookeeper中序列化和反序列化该元数据
17  *                 )是事务元数据集合,用户从数据源分发批次
18  * @author: hadoop
19  * @date: 2017年8月4日 下午7:46:11
20  *
21  */
22 public class TransSpout implements ITransactionalSpout<MyMata>{
23     private Map<Long,String> map = new HashMap<Long,String>();
24     @Override
25     public void declareOutputFields(OutputFieldsDeclarer arg0) {
26         arg0.declare(new Fields("atemptId","message"));
27     }
28 
29     @Override
30     public Map<String, Object> getComponentConfiguration() {
31         return null;
32     }
33 
34     /**
35      * 告诉storm用来协调生成批次的类，
36      */
37     @Override
38     public backtype.storm.transactional.ITransactionalSpout.Coordinator<MyMata> getCoordinator(Map arg0,
39             TopologyContext arg1) {
40         // TODO Auto-generated method stub
41         return new MyCoordinator();
42     }
43 
44     /**
45      * 负责读取批次并把它们分发到拓扑中的数据流组
46      */
47     @SuppressWarnings("unchecked")
48     @Override
49     public backtype.storm.transactional.ITransactionalSpout.Emitter<MyMata> getEmitter(Map arg0, TopologyContext arg1) {
50         return new MyEmitter(CreateData.getMapData(100L));
51     }
52 
53 }

package com.dxss.storm.transaction;

import java.math.BigInteger;

import backtype.storm.transactional.ITransactionalSpout.Coordinator;
import backtype.storm.utils.Utils;

public class MyCoordinator implements Coordinator<MyMata>{
    private static final Integer BATCH_NUM = 10;

    @Override
    public void close() {
    }

    /**
     * 启动事务，生成元数据。接收
     *  txid：storm生成的事务ID，作为批次的唯一性标识
     *  arg1: 协调器生成的前一个事务元数据对象
     *  元组一经返回，storm把myMata和txid一起保存在zookeeper,确保一旦发生故障，storm可以利用分发器重新发送批次
     */
    @Override
    public MyMata initializeTransaction(BigInteger txid, MyMata arg1) {
        // arg0 : 事务ID ，启动事务ID，默认从0开始
        // arg1 : 上一个元数据的位置
        long beginPoint = 0L;
        if (arg1 == null){
            beginPoint = 0;
        }else{
            beginPoint = arg1.getBeginPoint() + arg1.getNum();
        }
        MyMata myMata = new MyMata();
        myMata.setBeginPoint(beginPoint);
        myMata.setNum(BATCH_NUM);
        System.err.println("启动一个事务："+ myMata.toString());
        return myMata ;
    }

    /**
     * 使用它来确认数据源已经就绪并可读取
     */
    @Override
    public boolean isReady() {
        Utils.sleep(2000);
        return true;
    }

}

package com.dxss.storm.transaction;

import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;

import backtype.storm.coordination.BatchOutputCollector;
import backtype.storm.transactional.ITransactionalSpout.Emitter;
import backtype.storm.transactional.TransactionAttempt;
import backtype.storm.tuple.Values;

/**
 * 
 * @ClassName: MyEmitter
分发器，从数据源读取数据，并从数据流组发送数据，分发器可以为相同的事务id和事务元数据发送相同的批次，如果出现故障，storm
发送相同的事务id和事务元数据，并确保批次已经重复过了，storm会在TransactionAttempt对象里为
这样就知道批次已经重复过了
p
 * @date: 2017年8月5日 下午2:19:30
 *
 */
public class MyEmitter implements Emitter<MyMata>{

    private Map<Long,String> map = null;
    
    
    public MyEmitter(Map<Long, String> map2) {
        this.map = map2;
    }

    @Override
    public void cleanupBefore(BigInteger arg0) {
        // TODO Auto-generated method stub
        
    }

    @Override
    public void close() {
        // TODO Auto-generated method stub
        
    }

    /**
     *  从传入的元数据对象中获取原始消息进行发送，利用传入的MyMata元数据对象从map(消息源)中获取数据进行发送，同时增加map
     *  维持的已读消息数
     */
    @Override
    public void emitBatch(TransactionAttempt arg0, MyMata arg1, BatchOutputCollector arg2) {
        // 从某一位置获取原始消息进行发送
        long beginPoint = arg1.getBeginPoint();
        int num = arg1.getNum();
        for (long i=beginPoint;i<(num+beginPoint);i++){
            if (this.map.get(i)==null){
                break;
            }
            arg2.emit(new Values(arg0,this.map.get(i)));
        }
    }

}

package com.dxss.storm.transaction;

import java.util.Map;

import backtype.storm.coordination.BatchOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseTransactionalBolt;
import backtype.storm.transactional.TransactionAttempt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

@SuppressWarnings("serial")
public class TransBolt extends BaseTransactionalBolt{
    private BatchOutputCollector collector;
    private TransactionAttempt arg3 = null;
    private Integer count = 0;
    
    
    /**
     * execute 操作接收到的元组，但是不会分发新的元组，批次执行完成后，storm会调用finishBatch方法
     */
    @Override
    public void execute(Tuple arg0) {
        arg3 = (TransactionAttempt) arg0.getValue(0);
        System.out.println("MyTransactionBolt类中execute方法的事务ID是:"+ arg3.getTransactionId()
            + " attemptid"+arg3.getAttemptId()
                );
        String message = arg0.getString(1);
        if (message != null){
            count ++;
        }
    }

    /**
     * 批次执行完成后，storm会调用finishBatch方法，可以多线程并发处理，有几条线程，每条线程的批次处理完成后，就会调用该方法
     */
    @Override
    public void finishBatch() {
        System.err.println("FinishBatch的执行次数为：" + count);
        this.collector.emit(new Values(this.arg3,count));
    }

    
    @Override
    public void prepare(Map arg0, TopologyContext arg1, BatchOutputCollector arg2, TransactionAttempt arg3) {
        System.err.println("MyTransactionBolt类中prepare方法的事务ID是:"+ arg3.getAttemptId());
        this.arg3 = arg3;
        this.collector = arg2;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer arg0) {
        arg0.declare(new Fields("tx","count"));
        
    }

}

package com.dxss.storm.transaction;

import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;

import backtype.storm.coordination.BatchOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseTransactionalBolt;
import backtype.storm.transactional.ICommitter;
import backtype.storm.transactional.TransactionAttempt;
import backtype.storm.tuple.Tuple;

@SuppressWarnings("serial")
public class MyCommit extends BaseTransactionalBolt implements ICommitter{

    private static Map<String,DbValue> map = new HashMap<String,DbValue>();
    private Integer num = 0;
    private static final String GLOBAL_KEY = "GLOBAL_KEY";
    private TransactionAttempt arg3 ;
    
    class DbValue{
        private BigInteger txid ;
        int count = 0;
        public BigInteger getTxid() {
            return txid;
        }
        public void setTxid(BigInteger txid) {
            this.txid = txid;
        }
        public int getCount() {
            return count;
        }
        public void setCount(int count) {
            this.count = count;
        }
        
    }
    
    @Override
    public void execute(Tuple arg0) {
        this.num += arg0.getInteger(1);
    }

    /**
     * finishBatch 在提交就绪时执行，这一点发生在所有事务都已经成功提交之后，另外，finishBatch是顺序执行的，如果
     * 同时有事务ID1和事务ID2两个事务同时执行，只有在ID1没有差错的执行了finishBatch方法之后，ID2才会执行该方法
     */
    @SuppressWarnings("static-access")
    @Override
    public void finishBatch() {
        DbValue dbValue = this.map.get(GLOBAL_KEY);
        DbValue newDbValue = null;
        if (dbValue == null || !dbValue.txid.equals(this.arg3.getAttemptId())){
            //更新数据库
            newDbValue = new DbValue();
            newDbValue.txid = this.arg3.getTransactionId();
            if (dbValue == null){
                newDbValue.count = num;
            }else{
                newDbValue.count = dbValue.count + num;
            }
            this.map.put(GLOBAL_KEY, newDbValue);
        }else{
            newDbValue = dbValue;
        }
        System.err.println("total===================:"+this.map.get(GLOBAL_KEY).getCount());
    }

    @Override
    public void prepare(Map arg0, TopologyContext arg1, BatchOutputCollector arg2, TransactionAttempt arg3) {
        this.arg3 = arg3;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer arg0) {
    }

}