transactionmanager.java

java pos,你可以直接编译运行,

/*
 * Copyright (c) 2004 jPOS.org 
 *
 * See terms of license at http://jpos.org/license.html
 *
 */
package org.jpos.q2.transaction;

import java.io.Serializable;
import java.util.Map;
import java.util.List;
import java.util.HashMap;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.StringTokenizer;
import org.jpos.util.NameRegistrar;
import org.jdom.Element;
import org.jpos.space.Space;
import org.jpos.space.JDBMSpace;
import org.jpos.space.SpaceFactory;
import org.jpos.space.SpaceUtil;
import org.jpos.q2.QFactory;
import org.jpos.q2.QBeanSupport;
import org.jpos.core.ConfigurationException;
import org.jpos.util.Logger;
import org.jpos.util.LogEvent;
import org.jpos.transaction.TransactionConstants;
import org.jpos.transaction.TransactionParticipant;
import org.jpos.transaction.GroupSelector;
import org.jpos.transaction.AbortParticipant;
import org.jpos.transaction.ContextRecovery;

public class TransactionManager 
    extends QBeanSupport 
    implements Runnable, TransactionConstants, TransactionManagerMBean
{
    Space sp;
    Space psp;
    String queue;
    String tailLock;
    Map groups;
    long head, tail, lastGC;
    public static final String  HEAD       = "$HEAD";
    public static final String  TAIL       = "$TAIL";
    public static final String  CONTEXT    = "$CONTEXT.";
    public static final String  STATE      = "$STATE.";
    public static final String  GROUPS     = "$GROUPS.";
    public static final String  TAILLOCK   = "$TAILLOCK";
    public static final Integer PREPARING  = new Integer (0);
    public static final Integer COMMITTING = new Integer (1);
    public static final Integer DONE       = new Integer (2);
    public static final String  DEFAULT_GROUP = "";
    public static final long    MAX_PARTICIPANTS = 1000;  // loop prevention

    public void initService () throws ConfigurationException {
        queue = cfg.get ("queue", null);
        if (queue == null)
            throw new ConfigurationException ("queue property not specified");
        sp   = SpaceFactory.getSpace (cfg.get ("space"));
        psp  = SpaceFactory.getSpace (cfg.get ("persistent-space"));
        tail = initCounter (TAIL, cfg.getLong ("initial-tail", 1));
        head = Math.max (initCounter (HEAD, tail), tail);
        tailLock = TAILLOCK + "." + Integer.toString (this.hashCode());
        initTailLock ();

        groups = new HashMap();
        initParticipants (getPersist());
    }
    public void startService () {
        NameRegistrar.register (getName (), this);
        recover ();
        long sessions = cfg.getLong ("sessions", 1);
        for (int i=0; i<sessions; i++) {
            Thread t = new Thread (this);
            t.setName (getName() + "-" + i);
            t.start ();
        }
    }
    public void stopService () {
        NameRegistrar.unregister (getName ());
        long sessions = cfg.getLong ("sessions", 1);
        for (int i=0; i<sessions; i++)
            sp.out (queue, this, 60*1000);
    }
    public void queue (Serializable context) {
        sp.out (queue, context);
    }
    public void run () {
        long id;
        String threadName = Thread.currentThread().getName();
        getLog().info (threadName + " start");
        while (running()) {
            try {
                Object obj = sp.in (queue);
                if (obj == this)
                    continue;   // stopService ``hack''
                if (!(obj instanceof Serializable)) {
                    getLog().error (
                        "non serializable '" + obj.getClass().getName()
                       + "' on queue '" + queue + "'"
                    );
                    continue;
                }
                id = nextId ();
                List members = new ArrayList ();
                Serializable context = (Serializable) obj;
                snapshot (id, context, PREPARING);
                int action = prepare (id, context, members);
                switch (action) {
                    case PREPARED:
                        setState (id, COMMITTING);
                        commit (id, context, members, false);
                        break;
                    case ABORTED:
                        abort (id, context, members, false);
                        break;
                    case NO_JOIN:
                        break;
                }
                snapshot (id, null, DONE);
                if (id == tail) {
                    checkTail ();
                }
            } catch (Throwable t) {
                getLog().fatal (t); // should never happen
            } 
        }
        getLog().info (threadName + " stop");
    }

    public long getTail () {
        return tail;
    }
    public long getHead () {
        return head;
    }
    protected void commit 
        (long id, Serializable context, List members, boolean recover) 
    {
        Iterator iter = members.iterator();
        while (iter.hasNext ()) {
            TransactionParticipant p = (TransactionParticipant) iter.next();
            if (recover && p instanceof ContextRecovery)
                context = ((ContextRecovery) p).recover (id, context, true);
            commit (p, id, context);
        }
    }
    protected void abort 
        (long id, Serializable context, List members, boolean recover) 
    {
        Iterator iter = members.iterator();
        while (iter.hasNext ()) {
            TransactionParticipant p = (TransactionParticipant) iter.next();
            if (recover && p instanceof ContextRecovery)
                context = ((ContextRecovery) p).recover (id, context, false);
            abort (p, id, context);
        }
    }
    protected int prepareForAbort
        (TransactionParticipant p, long id, Serializable context) 
    {
        try {
            if (p instanceof AbortParticipant)
                return ((AbortParticipant)p).prepareForAbort (id, context);
        } catch (Throwable t) {
            getLog().warn ("PREPARE-FOR-ABORT: " + Long.toString (id), t);
        }
        return ABORTED | NO_JOIN;
    }
    protected int prepare 
        (TransactionParticipant p, long id, Serializable context) 
    {
        try {
            return p.prepare (id, context);
        } catch (Throwable t) {
            getLog().warn ("PREPARE: " + Long.toString (id), t);
        }
        return ABORTED;
    }
    protected void commit 
        (TransactionParticipant p, long id, Serializable context) 
    {
        try {
            p.commit (id, context);
        } catch (Throwable t) {
            getLog().warn ("COMMIT: " + Long.toString (id), t);
        }
    }
    protected void abort 
        (TransactionParticipant p, long id, Serializable context) 
    {
        try {
            p.abort (id, context);
        } catch (Throwable t) {
            getLog().warn ("ABORT: " + Long.toString (id), t);
        }
    }
    protected int prepare (long id, Serializable context, List members) {
        boolean abort = false;
        Iterator iter = getParticipants (DEFAULT_GROUP).iterator();
        for (int i=0; iter.hasNext (); i++) {
            int action = 0;
            if (i > MAX_PARTICIPANTS) {
                getLog().warn (
                    "loop detected - transaction " +id + " aborted."
                );
                return ABORTED;
            }
            TransactionParticipant p = (TransactionParticipant) iter.next();
            if (abort) {
                action = prepareForAbort (p, id, context);
            } else {
                action = prepare (p, id, context);
                abort  = (action & PREPARED) == ABORTED;
            }
            if ((action & READONLY) == 0) {
                snapshot (id, context);
            }
            if ((action & NO_JOIN) == 0) {
                members.add (p);
            }
            if (p instanceof GroupSelector) {
                String groupName = ((GroupSelector)p).select (id, context);
                if (groupName != null) {
                    StringTokenizer st = new StringTokenizer (groupName, " ,");