Java同步代码转异步代码

2019-11-17 04:02:57

字体：大中小

来源：转载

供稿：网友

java同步代码转异步代码
Kagula

2009-11-18

摘要
Thread   Thread Group

概要
   利用Java源码，介绍同步代码转异步代码框架的使用，下半部份附框架源代码，阅读本文之前可以参考《Java多线程_编程模型研究》http://blog.csdn.net/lee353086/archive/2008/01/10/2033587.aspx

正文
TestThreads.java源码用来示范异步调用

view plaincopy to clipboardPRint?
import kagula.multithread.MsgManagement;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class testThreads {
    private static Log log = LogFactory.getLog(testThreads.class);

    static public void main(String args[])
    {
        log.debug("-Test thread.begin!");

        try
        {
            Object lockMain=new Object(); //用来锁住主线程，可以用来等待branch thread返回运算结果。
            MsgManagement mm=new MsgManagement("MsgProc1",lockMain);  //MsgManagement的实例化
            String sTN =mm.invoke("t_main",new Class[]  {String.class},new Object[] {new String("FirstT")});
            String sTN2 =mm.invoke("t_main",new Class[]  {String.class},new Object[] {new String("SecondT")});

            log.debug("sTN="+sTN+" sTN2="+sTN2);

            if(mm.activeCount()>0)  {
                mm.setMaxPriority(Thread.MAX_PRIORITY);

                synchronized(lockMain)
                {
                  lockMain.wait(100);
                }
                mm.interrupt();
            }

            log.debug("mm.getResult(sTN) "+mm.getResult(sTN));
            log.debug("mm.getResult(sTN2) "+mm.getResult(sTN2));
        }catch(Exception ex)
        {
            ex.printStackTrace();
        }

        log.debug("-Test thread.end!");
    }
}
import kagula.multithread.MsgManagement;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class testThreads {
private static Log log = LogFactory.getLog(testThreads.class);

static public void main(String args[])
{
log.debug("-Test thread.begin!");

try
{
Object lockMain=new Object(); //用来锁住主线程，可以用来等待branch thread返回运算结果。
MsgManagement mm=new MsgManagement("MsgProc1",lockMain);  //MsgManagement的实例化
String sTN =mm.invoke("t_main",new Class[]  {String.class},new Object[] {new String("FirstT")});
String sTN2 =mm.invoke("t_main",new Class[]  {String.class},new Object[] {new String("SecondT")});

log.debug("sTN="+sTN+" sTN2="+sTN2);

if(mm.activeCount()>0)  {
mm.setMaxPriority(Thread.MAX_PRIORITY);

synchronized(lockMain)
{
      lockMain.wait(100);
    }
mm.interrupt();
}

log.debug("mm.getResult(sTN) "+mm.getResult(sTN));
log.debug("mm.getResult(sTN2) "+mm.getResult(sTN2));
}catch(Exception ex)
{
ex.printStackTrace();
}

log.debug("-Test thread.end!");
}
}


MsgManagement.java  Original Thread.java源码，是异步框架

view plaincopy to clipboardprint?
package kagula.multithread;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
* 线程管理器
*
* Requirement: JDK1.6.x or above
*
* @author Jun Li
*
*/
public class MsgManagement extends ThreadGroup{
    protected static Log log = LogFactory.getLog(MsgManagement.class);

    private int nID=1;  //auxiliary thread counter flag

    private Map mapResults=Collections.synchronizedMap(new HashMap());  //store threads processing result.
    private Object lockMain;

//  -----------------below methods using only by main callers.begin-----------------
    public MsgManagement(String strGroupName,Object pLock)
    {
        //Initialization the attributes
        super(strGroupName);
        lockMain=pLock;
    }

    /**
     * After send processing request,will be return soon.
     * @param msg
     * Message
     * @param nParam
     * First Parameter
     * @param secondParam
     * Second Parameter
     * @return
     */
    public synchronized String invoke(String pMethodName,Class pClass[],Object pObject[]) throws Exception
    {
        return invoke(pMethodName,pClass,pObject,lockMain);
    }

    public synchronized String invoke(String pMethodName,Class pClass[],Object pObject[],Object pLock) throws Exception
    {
        String strThreadName=new String("MsgManagement's No."+nID+" thread");
        MyThread thread=new MyThread(this,strThreadName);
        thread.setLockMain(pLock);
        thread.setMethodName(pMethodName);
        thread.setClassParams(pClass);
        thread.setObjParams(pObject);
        thread.start();
        nID++;
        return strThreadName;
    }

    /**
     * Get result by thread name
     * @param pThreadName
     * @return
     */
    public Object getResult(String pThreadName)
    {
        return mapResults.get(pThreadName);
    }
//  -----------------above methods using only by main callers.end-----------------

    //-----------------below methods using only by threads.begin-----------------
    public void setNResult(String pName,Object pObj)
    {
        mapResults.put(pName, pObj);
    }
//  -----------------above methods using only by threads.begin-----------------
}
package kagula.multithread;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
* 线程管理器
*
* Requirement: JDK1.6.x or above
*
* @author Jun Li
*
*/
public class MsgManagement extends ThreadGroup{
protected static Log log = LogFactory.getLog(MsgManagement.class);

private int nID=1;  //auxiliary thread counter flag

    private Map mapResults=Collections.synchronizedMap(new HashMap());  //store threads processing result.
    private Object lockMain;

//  -----------------below methods using only by main callers.begin-----------------
    public MsgManagement(String strGroupName,Object pLock)
    {
     //Initialization the attributes
     super(strGroupName);
     lockMain=pLock;
    }

    /**
     * After send processing request,will be return soon.
     * @param msg
     * Message
     * @param nParam
     * First Parameter
     * @param secondParam
     * Second Parameter
     * @return
     */
public synchronized String invoke(String pMethodName,Class pClass[],Object pObject[]) throws Exception
{
return invoke(pMethodName,pClass,pObject,lockMain);
}

    public synchronized String invoke(String pMethodName,Class pClass[],Object pObject[],Object pLock) throws Exception
    {
String strThreadName=new String("MsgManagement's No."+nID+" thread");
MyThread thread=new MyThread(this,strThreadName);
thread.setLockMain(pLock);
thread.setMethodName(pMethodName);
thread.setClassParams(pClass);
thread.setObjParams(pObject);
thread.start();
nID++;
return strThreadName;
    }

    /**
     * Get result by thread name
     * @param pThreadName
     * @return
     */
    public Object getResult(String pThreadName)
    {
     return mapResults.get(pThreadName);
    }
//  -----------------above methods using only by main callers.end-----------------

    //-----------------below methods using only by threads.begin-----------------
public void setNResult(String pName,Object pObj)
{
mapResults.put(pName, pObj);
}
// -----------------above methods using only by threads.begin-----------------
}

view plaincopy to clipboardprint?
package kagula.multithread;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
* Your object need derived from it to implement asynchronous method!
* @author Lijun
*
*/
public abstract class OriginalThread extends Thread{
    protected static Log log = LogFactory.getLog(OriginalThread.class);
    protected String MethodName=new String("default");
    protected Object objParams[];
    protected Class  classParams[];
    protected Object lockMain;
    public OriginalThread(ThreadGroup tg,String name)
    {
        super(tg,name);
    }

    public void run()
    {
        log.debug("["+getName()+"] instance of OriginalThread: Before Invoke the method ["+MethodName+"]");
        Method method=null;
        Object objR=null;
        try {
            method=this.getClass().getMethod(MethodName, classParams);
        }catch(NoSuchMethodException e)
        {
            log.error(e.toString());
            return;
        }
        if(method!=null)
        {
            try {
                objR=method.invoke(this, objParams);
            }catch(Illegal accessException e)
            {
                log.error(e.toString());
                return;
            }catch(InvocationTargetException e)
            {
                log.error(e.toString());
                return;
            }catch(IllegalArgumentException e)
            {
                log.error(e);
                return;
            }
            catch(Exception e)
            {
                log.error(e);
                return;
            }
        }
        log.debug("["+getName()+"] instance of OriginalThread: After Invoke the method ["+MethodName+"] return value is ["+objR+"]");

        setNResult(objR);  //put setNResult before setNFlag,because assure after nFlag==SUCCESS,caller will get proper result!
        synchronized(lockMain) {
            lockMain.notifyAll();
        }
    }

    //---------------------below methods only should be used by MsgManagement---------------------
    private void setNResult(Object pObj)
    {
        MsgManagement mm=(MsgManagement)this.getThreadGroup();
        mm.setNResult(this.getName(), pObj);
    }

    public void setMethodName(String pMethodName) {
        this.MethodName = pMethodName;
    }

    public void setObjParams(Object[] objParams) {
        this.objParams = objParams;
    }

    public void setClassParams(Class[] classParams) {
        this.classParams = classParams;
    }

    public void setLockMain(Object pLock)
    {
        lockMain=pLock;
    }
}
package kagula.multithread;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
* Your object need derived from it to implement asynchronous method!
* @author Lijun
*
*/
public abstract class OriginalThread extends Thread{
protected static Log log = LogFactory.getLog(OriginalThread.class);
protected String MethodName=new String("default");
protected Object objParams[];
protected Class  classParams[];
protected Object lockMain;
public OriginalThread(ThreadGroup tg,String name)
{
super(tg,name);
}

public void run()
{
log.debug("["+getName()+"] instance of OriginalThread: Before Invoke the method ["+MethodName+"]");
Method method=null;
Object objR=null;
    try {
     method=this.getClass().getMethod(MethodName, classParams);
    }catch(NoSuchMethodException e)
    {
     log.error(e.toString());
     return;
    }
    if(method!=null)
    {
     try {
     objR=method.invoke(this, objParams);
     }catch(IllegalAccessException e)
     {
     log.error(e.toString());
     return;
     }catch(InvocationTargetException e)
     {
     log.error(e.toString());
     return;
     }catch(IllegalArgumentException e)
     {
     log.error(e);
     return;
     }
     catch(Exception e)
     {
     log.error(e);
     return;
     }
    }
    log.debug("["+getName()+"] instance of OriginalThread: After Invoke the method ["+MethodName+"] return value is ["+objR+"]");

setNResult(objR);  //put setNResult before setNFlag,because assure after nFlag==SUCCESS,caller will get proper result!
synchronized(lockMain) {
lockMain.notifyAll();
}
}

//---------------------below methods only should be used by MsgManagement---------------------
private void setNResult(Object pObj)
{
MsgManagement mm=(MsgManagement)this.getThreadGroup();
mm.setNResult(this.getName(), pObj);
}

public void setMethodName(String pMethodName) {
this.MethodName = pMethodName;
}

public void setObjParams(Object[] objParams) {
this.objParams = objParams;
}

public void setClassParams(Class[] classParams) {
this.classParams = classParams;
}

public void setLockMain(Object pLock)
{
lockMain=pLock;
}
}

MyThread.java是，你同步代码存放的地方

view plaincopy to clipboardprint?
package kagula.multithread;

import org.apache.commons.logging.LogFactory;

public class MyThread extends OriginalThread{
    public MyThread(ThreadGroup tg,String name)
    {
        super(tg,name);

        log=LogFactory.getLog(this.getClass());
    }

    public String t_main(String sT)
    {
        final long nWaitTime=1*1*5*1000;  //H*M*S,once,more than nWaitTime hours

        String s=null;;

        try{
            log.debug("before sleep "+sT);
            sleep(nWaitTime);
            log.debug("after sleep "+sT);

            s="I will return. "+sT;
            return s;
        }catch(InterruptedException e)
        {
            log.error("kagula("+getId()+"):"+e.toString());
        }
        return s;
    }
}

上一篇：免费的java虚拟主机空间 google的app engine

下一篇：java领域即时通信的解决方案一(openfire+spark+smack)