java actor模型和消息传递实现分析

Actor模型是一种基于协程的消息传递模型，在并行计算和并发的消息传递中有很好的性能表现。一般的actor模块框架提供了超轻量级的线程和工具，可以在这些线程之间进行快速、安全、零复制的消息传递。在elang，ruby，lua等语言中都是直接在VM级别支持协程，VM帮你做context的保存和恢复。而在java中，却没有内置actor模型实现，但是有几个开源框架也模拟了actor模型的实现。

 

基于 actor 的系统 通过实现一种消息传递 模式，使并行处理更容易编码。在此模式中，系统中的每个 actor 都可接收消息；执行该消息所表示的操作；然后将消息发送给其他 actor（包括它们自己）以执行复杂的操作序列。actor 之间的所有消息是异步的，这意味着发送者会在收到任何回复之前继续进行处理。因此，一个 actor 可能终生都陷入接收和处理消息的无限循环中。

当使用多个 actor 时，独立的活动可轻松分配到多个可并行执行消息的线程上（进而分配在多个处理器上）。一般而言，每个 actor 都在一个独立线程上处理消息。一些 actor 系统静态地向 actor 分配线程；而其他系统（比如本文中介绍的系统）则会动态地分配它们。

 

 

下面我们会分析下java中的一个actor模型框架的实现：

 

我们先看下elang中的actor模型的实现：

 

以Erlang为例子，介绍一下简单的Actor模型

 1.首先建立一个Actor，在erlang中，起一个进程（这个是erlang虚拟机进程，跟os进程不同),这个进程就是actor了，可以用来接收和发送各种消息了

 Pid = spawn(Mod,func,Args) %起一个进程

 2.处理收到的消息

func()->

    receive

{From,Msg}-> %收到一个消息

%%do something

func();

 3.要对这个actor发送消息，也非常简单

  Pid ! {From,Msg}

 

 

ujavaactor框架：

 

下面摘自ibm developer的一段介绍

 

μJavaActors 是 actor 系统的一个简单的 Java 实现。只有 1,200 行代码，μJavaActors 虽然很小，但很强大。在下面的练习中，您将学习如何使用 μJavaActors 动态地创建和管理 actor，将消息传送给它们。

μJavaActors 围绕 3 个核心界面而构建：

• 消息 是在 actor 之间发送的消息。Message 是 3 个（可选的）值和一些行为的容器：
  • source 是发送 actor。
  • subject 是定义消息含义的字符串（也称为命令）。
  • data 是消息的任何参数数据；通常是一个映射、列表或数组。参数可以是要处理和/或其他 actor 要与之交互的数据。
  • subjectMatches() 检查消息主题是否与字符串或正则表达式匹配。
  μJavaActors 包的默认消息类是 DefaultMessage。
• ActorManager 是一个 actor 管理器。它负责向 actor 分配线程（进而分配处理器）来处理消息。ActorManager 拥有以下关键行为或特征：
  • createActor() 创建一个 actor 并将它与此管理器相关联。
  • startActor() 启动一个 actor。
  • detachActor() 停止一个 actor 并将它与此管理器断开。
  • send()/broadcast() 将一条消息发送给一个 actor、一组 actor、一个类别中的任何 actor 或所有 actor。
  在大部分程序中，只有一个 ActorManager，但如果您希望管理多个线程和/或 actor 池，也可以有多个 ActorManager。此接口的默认实现是 DefaultActorManager。
• Actor 是一个执行单元，一次处理一条消息。Actor 具有以下关键行为或特征：
  • 每个 actor 有一个 name，该名称在每个 ActorManager 中必须是惟一的。
  • 每个 actor 属于一个 category；类别是一种向一组 actor 中的一个成员发送消息的方式。一个 actor 一次只能属于一个类别。
  • 只要 ActorManager 可以提供一个执行 actor 的线程，系统就会调用 receive()。为了保持最高效率，actor 应该迅速处理消息，而不要进入漫长的等待状态（比如等待人为输入）。
  • willReceive() 允许 actor 过滤潜在的消息主题。
  • peek() 允许该 actor 和其他 actor 查看是否存在挂起的消息（或许是为了选择主题）。
  • remove() 允许该 actor 和其他 actor 删除或取消任何尚未处理的消息。
  • getMessageCount() 允许该 actor 和其他 actor 获取挂起的消息数量。
  • getMaxMessageCount() 允许 actor 限制支持的挂起消息数量；此方法可用于预防不受控制地发送。
  大部分程序都有许多 actor，这些 actor 常常具有不同的类型。actor 可在程序启动时创建或在程序执行时创建（和销毁）。本文中的 actor 包 包含一个名为 AbstractActor 的抽象类，actor 实现基于该类。

 

 

 

public abstract class AbstractActor extends Utils implements Actor {
	public static final int DEFAULT_MAX_MESSAGES = 100;
	protected DefaultActorManager manager;

	public ActorManager getManager() {
		return manager;
	}

	public void setManager(DefaultActorManager manager) {
		if (this.manager != null && manager != null) {
			throw new IllegalStateException(
					"cannot change manager of attached actor");
		}
		this.manager = manager;
	}

	protected String name;

	@Override
	public String getName() {
		return name;
	}

	@Override
	public void setName(String name) {
		if (manager != null) {
			throw new IllegalStateException("cannot change name if manager set");
		}
		this.name = name;
	}

	protected String category = DEFAULT_CATEGORY;

	@Override
	public String getCategory() {
		return category;
	}

	@Override
	public void setCategory(String category) {
		this.category = category;
	}

	/**
	 * Process a message conditionally. If testMessage() returns null no message
	 * will be consumed.
	 * 
	 * @see AbstractActor#testMessage()
	 */
	@Override
	public boolean receive() {
		Message m = testMessage();
		boolean res = m != null;
		if (res) {
			boolean f = remove(m);
			if (!f) {
				logger.warning("receive message not removed: %s", m);
			}
			DefaultMessage dm = (DefaultMessage) m;
			try {
				dm.fireMessageListeners(new MessageEvent(this, dm, MessageEvent.MessageStatus.DELIVERED));
				//logger.trace("receive %s processing %s", this.getName(), m);
				loopBody(m);
				dm.fireMessageListeners(new MessageEvent(this, dm, MessageEvent.MessageStatus.COMPLETED));
			} catch (Exception e) {
				dm.fireMessageListeners(new MessageEvent(this, dm, MessageEvent.MessageStatus.FAILED));
				logger.error("loop exception", e);
			}
		}
		manager.awaitMessage(this);
		return res;
	}

	/**
	 * Test to see if a message should be processed. Subclasses should override
	 */
	@Override
	public boolean willReceive(String subject) {
		return !isEmpty(subject); // default receive all subjects
	}

	/** Test the current message. Default action is to accept all. */
	protected Message testMessage() {
		return getMatch(null, false);
	}

	/** Process the accepted subject. */
	abstract protected void loopBody(Message m);

	/** Test a message against a defined subject pattern. */
	protected DefaultMessage getMatch(String subject, boolean isRegExpr) {
		DefaultMessage res = null;
		synchronized (messages) {
			res = (DefaultMessage) peekNext(subject, isRegExpr);
		}
		return res;
	}

	protected List<DefaultMessage> messages = new LinkedList<DefaultMessage>();

	public DefaultMessage[] getMessages() {
		return messages.toArray(new DefaultMessage[messages.size()]);
	}

	@Override
	public int getMessageCount() {
		synchronized (messages) {
			return messages.size();
		}
	}

	/**
	 * Limit the number of messages that can be received.  Subclasses should override.
	 */
	@Override
	public int getMaxMessageCount() {
		return DEFAULT_MAX_MESSAGES;
	}

	/** Queue a messaged to be processed later. */
	public void addMessage(DefaultMessage message) {
		if (message != null) {
			synchronized (messages) {
				if (messages.size() < getMaxMessageCount()) {
					messages.add(message);
					// messages.notifyAll();
				} else {
					throw new IllegalStateException("too many messages, cannot add");
				}
			}
		} 
	}

	@Override
	public Message peekNext() {
		return peekNext(null);
	}

	@Override
	public Message peekNext(String subject) {
		return peekNext(subject, false);
	}

	/** 
	 * See if a message exists that meets the selection criteria. 
	 **/
	@Override
	public Message peekNext(String subject, boolean isRegExpr) {
		Message res = null;
		if (isActive) {
			Pattern p = subject != null ? (isRegExpr ? Pattern.compile(subject)
					: null) : null;
			long now = new Date().getTime();
			synchronized (messages) {
				for (DefaultMessage m : messages) {
					if (m.getDelayUntil() <= now) {
						boolean match = subject == null
								|| (isRegExpr ? m.subjectMatches(p) : m
										.subjectMatches(subject));
						if (match) {
							res = m;
							break;
						}
					}
				}
			}
		}
		// logger.trace("peekNext %s, %b: %s", subject, isRegExpr, res);
		return res;
	}

	@Override
	public boolean remove(Message message) {
		synchronized (messages) {
			return messages.remove(message);
		}
	}

	protected boolean isActive;

	public boolean isActive() {
		return isActive;
	}

	@Override
	public void activate() {
		isActive = true;
	}

	@Override
	public void deactivate() {
		isActive = false;
	}

	/** Do startup processing. */
	protected void runBody() {
		DefaultMessage m = new DefaultMessage("init");
		getManager().send(m, null, this);
	}

	@Override
	public void run() {
		runBody();
		((DefaultActorManager) getManager()).awaitMessage(this);
	}

	protected boolean hasThread;

	public boolean getHasThread() {
		return hasThread;
	}

	protected void setHasThread(boolean hasThread) {
		this.hasThread = hasThread;
	}

	@Override
	public String toString() {
		return getClass().getSimpleName() + "[" + bodyString() + "]";
	}

	protected String bodyString() {
		return "name=" + name + ", category=" + category + ", messages="
				+ messages.size();
	}

	volatile protected boolean shutdown;

	@Override
	public boolean isShutdown() {
		return shutdown;
	}

	@Override
	public void shutdown() {
		shutdown = true;
	}

	volatile protected boolean suspended;

	@Override
	public void setSuspended(boolean f) {
		suspended = f;
	}

	@Override
	public boolean isSuspended() {
		return suspended;
	}
}

 

 

我们需要继承AbstractActor抽象类，实现loopBody方法来处理消息内容。

其中receive方法是接收一条消息并进行处理，首先取出一条消息，从消息列表中删除，然后出发消息接收前的MessageEvent.MessageStatus.DELIVERED事件，我们可以实现MessageListener接口来监听消息处理的时间，然后通过抽象方法loopBody来处理消息，最后触发消息接收后的MessageEvent.MessageStatus.COMPLETED事件。

peekNext只是取出消息。

其中isActive表示是否actor已经激活，hasThread表示是否已经分配线程在执行，shutdown是否已经关闭，suspended是否已经挂起。

 

我们看下defaultMessage消息体：

 

public class DefaultMessage extends Utils implements Message {

......

	protected Actor source;


	protected String subject;


	protected Object data;


	protected List<MessageListener> listeners = new LinkedList<MessageListener>();

	public void addMessageListener(MessageListener l) {
		if (!listeners.contains(l)) {
			listeners.add(l);
		}
	}

	public void removeMessageListener(MessageListener l) {
		listeners.remove(l);
	}
	
	public void fireMessageListeners(MessageEvent e) {
		for(MessageListener l : listeners) {
			l.onMessage(e);
		}
	}
	......
}

 

 可以给每个消息增加一些消息监听事件。

 

然后重点再看下DefaultActorManager的实现：

 

初始化：

初始化会创建ActorManager.properties读取配置的启动线程数参数，然后创建一个线程组，创建getThreadCount（）个ActorRunnable的线程（用来执行runnables保存的待执行actor队列中的任务），

最后还启动一个Counter的线程，用于计算每个调度时间（1秒）的执行actor数。

 

 

/**
	 * Get the default instance. Uses ActorManager.properties for configuration.
	 * 
	 * @return shared instance
	 */
	public static DefaultActorManager getDefaultInstance() {
		if (instance == null) {
			instance = new DefaultActorManager();
			Map<String, Object> options = null;
			// ConfigUtils configUtils = new ConfigUtils();
			// Properties p = configUtils
			// .loadProperties("ActorManager.properties");
			Properties p = new Properties();
			try {
				p.load(new FileInputStream("ActorManager.properties"));
			} catch (IOException e) {
				try {
					p.load(new FileInputStream("/resource/ActorManager.properties"));
				} catch (IOException e1) {
					logger.warning("DefaultActorManager: no configutration: " + e);
				}
			}
			if (!isEmpty(p)) {
				options = new HashMap<String, Object>();
				for (Object key : p.keySet()) {
					String skey = (String) key;
					options.put(skey, p.getProperty(skey));
				}
			}
			instance.initialize(options);
		}
		return instance;
	}

protected ThreadGroup threadGroup; 	protected static int groupCount;

	protected List<Thread> threads = new LinkedList<Thread>(); //保存线程池的线程列表
	/**
	 * Initialize this manager. Call only once.
	 * 
	 * @param options
	 *            map of options
	 */
	@Override
	public void initialize(Map<String, Object> options) {
		if (!initialized) {
			initialized = true;
			int count = getThreadCount(options);
			ThreadGroup tg = new ThreadGroup("ActorManager" + groupCount++);
			threadGroup = tg;
			for (int i = 0; i < count; i++) {
				createThread(i);
			}
			running = true;
			for (Thread t : threads) {
				// logger.trace("procesNextActor starting %s", t);
				t.start();
			}

			Thread Counter = new Thread(new Runnable() {
				@Override
				public void run() {
					while (running) {
						try {
							trendValue = sendCount - dispatchCount;
							// logger.trace("Counter thread: sc=%d, dc=%d, t=%d",
							// sendCount, dispatchCount, trendValue);
							lastSendCount = sendCount;
							sendCount = 0;
							updateLastDispatchCount();
							Thread.sleep(1000);
						} catch (InterruptedException e) {
							break;
						}
					}
					sendCount = lastSendCount = 0;
					clearDispatchCount();
				}
			});
			Counter.setDaemon(true);
			lastDispatchTime = lastSendTime = new Date().getTime();
			Counter.start();
		}
	}
	protected void createThread(int i) {
		addThread("actor" + i);
	}
/**
	 * Add a dynamic thread. 
	 * 
	 * @param name
	 * @return
	 */
	public Thread addThread(String name) {
		Thread t = null;
		synchronized (actors) {
			if (trunnables.containsKey(name)) {
				throw new IllegalStateException("already exists: " + name);
			}
			ActorRunnable r = new ActorRunnable();
			trunnables.put(name, r);
			t = new Thread(threadGroup, r, name);
			threads.add(t);
			//System.out.printf("addThread: %s", name);
		}
		t.setDaemon(true);
		t.setPriority(getThreadPriority());
		return t;
	}

 

 

 

 

 

	/** Configuration key for thread count. */
	public static final String ACTOR_THREAD_COUNT = "threadCount";

	protected Map<String, AbstractActor> actors = new LinkedHashMap<String, AbstractActor>();

	protected Map<String, AbstractActor> runnables = new LinkedHashMap<String, AbstractActor>();

	protected Map<String, AbstractActor> waiters = new LinkedHashMap<String, AbstractActor>();

 

 

actors存放所有的actor对象，runnables存放待发送消息（发送方）的actor， waiters存放等待（接收方接收消息）的actors。

 

 

创建一个actor实例：

 

	/**
	 * Create an actor and associate it with this manager.
	 * 
	 * @param clazz
	 *            the actor class
	 * @param the
	 *            actor name; must be unique
	 * @param options
	 *            actor options
	 */
	@Override
	public Actor createActor(Class<? extends Actor> clazz, String name, Map<String, Object> options) {
		AbstractActor a = null;
		synchronized (actors) {
			if (!actors.containsKey(name)) {
				try {
					a = (AbstractActor) clazz.newInstance();
					a.setName(name);
					a.setManager(this);
				} catch (Exception e) {
					throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(
							"mapped exception: " + e, e);
				}
			} else {
				throw new IllegalArgumentException("name already in use: " + name);
			}
		}
		return a;
	}

	/**
	 * Start an actor. Must have been created by this manager.
	 * 
	 * @param actor
	 *            the actor
	 */
	@Override
	public void startActor(Actor actor) {
		if (((AbstractActor) actor).getManager() != this) {
			throw new IllegalStateException("actor not owned by this manager");
		}
		String name = actor.getName();
		synchronized (actors) {
			if (actors.containsKey(name)) {
				throw new IllegalStateException("already started");
			}
			((AbstractActor) actor).shutdown = false;
			actors.put(name, (AbstractActor) actor);
			runnables.put(name, (AbstractActor) actor);
		}
		actor.activate();
	}

  启动一个actor实例会把actor放入actors队列和runnables等待执行队列（发送方发送消息）中，然后调用actor的activate（）方法做初始化。

 

 

我们再看看ActorRunnable线程池中的线程执行actor的过程：

 

actorRunnable.java是DefaultActorManager的内部类，可以用于访问DefaultActorManager的actor实例列表

 

procesNextActor 从runnables的待执行actor列表中取出第一个（fifo）actor，如果不为空，则执行该actor线程的run方法，发送消息，actor接口是继承了runable接口的，如果runnables列表为空，则从waiters列表中取出一个actor执行，接收消息，接收到消息，响应的增加dispatchCount的值，表示已经消费的消息数。如果procesNextActor 执行失败，则主线程等待100ms，否则循环执行。

 

 

/** public intended only for "friend" access. */
	public class ActorRunnable implements Runnable {
		public boolean hasThread;
		public AbstractActor actor;

		public void run() {
			// logger.trace("procesNextActor starting");
			int delay = 1;
			while (running) {
				try {
					if (!procesNextActor()) {
						// logger.trace("procesNextActor waiting on actor");
						// sleep(delay * 1000);
						synchronized (actors) {
							// TOOD: adjust this delay; possible parameter
							// we want to minizmize overhead (make bigger);
							// but it has a big impact on message processing
							// rate (makesmaller)
							// actors.wait(delay * 1000);
							actors.wait(100);
						}
						delay = Math.max(5, delay + 1);
					} else {
						delay = 1;
					}
				} catch (InterruptedException e) {
				} catch (Exception e) {
					logger.error("procesNextActor exception", e);
				}
			}
			// logger.trace("procesNextActor ended");
		}

		protected boolean procesNextActor() {
			boolean run = false, wait = false, res = false;
			actor = null;
			synchronized (actors) {
				for (String key : runnables.keySet()) {
					actor = runnables.remove(key);
					break;
				}
			}
			if (actor != null) {
				// first run never started
				run = true;
				actor.setHasThread(true);
				hasThread = true;
				try {
					actor.run();
				} finally {
					actor.setHasThread(false);
					hasThread = false;
				}
			} else {
				synchronized (actors) {
					for (String key : waiters.keySet()) {
						actor = waiters.remove(key);
						break;
					}
				}
				if (actor != null) {
					// then waiting for responses
					wait = true;
					actor.setHasThread(true);
					hasThread = true;
					try {
						res = actor.receive();
						if (res) {
							incDispatchCount();
						}
					} finally {
						actor.setHasThread(false);
						hasThread = false;
					}
				}
			}
			// if (!(!run && wait && !res) && a != null) {
			// logger.trace("procesNextActor %b/%b/%b: %s", run, wait, res, a);
			// }
			return run || res;
		}
	}

 

 

接下来就是调用发送消息的api了，send将消息message从from发送到to actor中，sentMessages保存了每个actor的消息列表，还有批量发送到多个actor中，send(Message message, Actor from, String category)是将消息发送到分类为category的actor中。broadcast是将消息发送到所有的actor中。

 

 

/**
	 * Send a message.
	 * 
	 * @param message
	 *            message to
	 * @param from
	 *            source actor
	 * @param to
	 *            target actor
	 * @return number of receiving actors
	 */
	@Override
	public int send(Message message, Actor from, Actor to) {
		int count = 0;
		if (message != null) {
			AbstractActor aa = (AbstractActor) to;
			if (aa != null) {
				if (!aa.isShutdown() && !aa.isSuspended() && aa.willReceive(message.getSubject())) {
					DefaultMessage xmessage = (DefaultMessage) ((DefaultMessage) message).assignSender(from);
					// logger.trace(" %s to %s", xmessage, to);
					aa.addMessage(xmessage);
					xmessage.fireMessageListeners(new MessageEvent(aa, xmessage, MessageEvent.MessageStatus.SENT));
					sendCount++;
					lastSendTime = new Date().getTime();
					if (recordSentMessages) {
						synchronized (sentMessages) {
							String aname = aa.getName();
							List<Message> l = sentMessages.get(aname);
							if (l == null) {
								l = new LinkedList<Message>();
								sentMessages.put(aname, l);
							}
							// keep from getting too big
							if (l.size() < 100) {
								l.add(xmessage);
							}
						}
					}
					count++;
					synchronized (actors) {
						actors.notifyAll();
					}
				}
			}
		}
		return count;
	}

	/**
	 * Send a message.
	 * 
	 * @param message
	 *            message to
	 * @param from
	 *            source actor
	 * @param to
	 *            target actors
	 * @return number of receiving actors
	 */
	@Override
	public int send(Message message, Actor from, Actor[] to) {
		int count = 0;
		for (Actor a : to) {
			count += send(message, from, a);
		}
		return count;
	}

	/**
	 * Send a message.
	 * 
	 * @param message
	 *            message to
	 * @param from
	 *            source actor
	 * @param to
	 *            target actors
	 * @return number of receiving actors
	 */
	@Override
	public int send(Message message, Actor from, Collection<Actor> to) {
		int count = 0;
		for (Actor a : to) {
			count += send(message, from, a);
		}
		return count;
	}

	/**
	 * Send a message.
	 * 
	 * @param message
	 *            message to
	 * @param from
	 *            source actor
	 * @param category
	 *            target actor category
	 * @return number of receiving actors
	 */
	@Override
	public int send(Message message, Actor from, String category) {
		int count = 0;
		Map<String, Actor> xactors = cloneActors();
		List<Actor> catMembers = new LinkedList<Actor>();
		for (String key : xactors.keySet()) {
			Actor to = xactors.get(key);
			if (category.equals(to.getCategory()) && (to.getMessageCount() < to.getMaxMessageCount())) {
				catMembers.add(to);
			}
		}
		// find an actor with lowest message count
		int min = Integer.MAX_VALUE;
		Actor amin = null;
		for (Actor a : catMembers) {
			int mcount = a.getMessageCount();
			if (mcount < min) {
				min = mcount;
				amin = a;
			}
		}
		if (amin != null) {
			count += send(message, from, amin);
			// } else {
			// throw new
			// IllegalStateException("no capable actors for category: " +
			// category);
		}
		return count;
	}

	/**
	 * Send a message to all actors.
	 * 
	 * @param message
	 *            message to
	 * @param from
	 *            source actor
	 * @return number of receiving actors
	 */
	@Override
	public int broadcast(Message message, Actor from) {
		int count = 0;
		Map<String, Actor> xactors = cloneActors();
		for (String key : xactors.keySet()) {
			Actor to = xactors.get(key);
			count += send(message, from, to);
		}
		return count;
	}

 

这里每次获取所有的actor列表都是采用clone方式复制当前的actor列表，如果长时间加锁，则会降低并发能力。

 

 

	protected Map<String, Actor> cloneActors() {
		Map<String, Actor> xactors;
		synchronized (actors) {
			xactors = new HashMap<String, Actor>(actors);
		}
		return xactors;
	}

 

当然这里还有termina所有的线程

 

 

/**
	 * Terminate processing and wait for all threads to stop.
	 */
	@Override
	public void terminateAndWait() {
		logger.trace("terminateAndWait waiting on termination of %d threads", threads.size());
		terminate();
		waitForThreads();
	}

	/**
	 * Wait for all threads to stop. Must have issued terminate.
	 */
	public void waitForThreads() {
		if (!terminated) {
			throw new IllegalStateException("not terminated");
		}
		for (Thread t : threads) {
			try {
				// logger.info("terminateAndWait waiting for %s...", t);
				t.join();
			} catch (InterruptedException e) {
				// logger.info("terminateAndWait interrupt");
			}
		}
	}

	boolean running, terminated;

	/**
	 * Terminate processing.
	 */
	@Override
	public void terminate() {
		terminated = true;
		running = false;
		for (Thread t : threads) {
			t.interrupt();
		}
		synchronized (actors) {
			for (String key : actors.keySet()) {
				actors.get(key).deactivate();
			}
		}
		sentMessages.clear();
		sendCount = lastSendCount = 0;
		clearDispatchCount();
	}

 

也可以挂起某个actor，直到有消息达到，这里就是讲actor加入waiters列表，注意上面send发送消息之后，会调用acoters.nofityAll唤醒等待线程

 

 

	/**
	 * Suspend an actor until it has a read message.
	 * 
	 * @param actor
	 *            receiving actor
	 */
	public void awaitMessage(AbstractActor actor) {
		synchronized (actors) {
			waiters.put(actor.getName(), actor);
			// actors.notifyAll();
			// logger.trace("awaitMessage waiters=%d: %s",waiters.size(), a);
		}
	}

 

startActor时才将actor加入runnables队列。

 

示例代码：

 

 

import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

import javax.swing.event.ChangeEvent;
import javax.swing.event.ChangeListener;

import com.ibm.actor.Actor;
import com.ibm.actor.DefaultActorManager;
import com.ibm.actor.DefaultMessage;
import com.ibm.actor.logging.DefaultLogger;
import com.ibm.actor.utils.Utils;

/** 
 * A set of runtime services for testing actors and a test case driver. 
 * 
 * @author BFEIGENB
 *
 */
public class DefaultActorTest extends Utils {

	public static final int MAX_IDLE_SECONDS = 10;

	// public static final int STEP_COUNT = 3 * 60;
	public static final int TEST_VALUE_COUNT = 1000; // TODO: make bigger

	public DefaultActorTest() {
		super();
	}

	private Map<String, Actor> testActors = new ConcurrentHashMap<String, Actor>();

	static Random rand = new Random();

	public static int nextInt(int limit) {
		return rand.nextInt(limit);
	}

	protected DefaultActorManager getManager() {
		DefaultActorManager am = actorManager != null ? actorManager : new DefaultActorManager();
		return am;
	}

	protected int stepCount = 120;

	public void setStepCount(int stepCount) {
		this.stepCount = stepCount;
	}

	public int getStepCount() {
		return stepCount;
	}

	protected int threadCount = 10;

	public int getThreadCount() {
		return threadCount;
	}

	public void setThreadCount(int threadCount) {
		this.threadCount = threadCount;
	}

	public void setTestActors(Map<String, Actor> testActors) {
		this.testActors = testActors;
	}

	public Map<String, Actor> getTestActors() {
		return testActors;
	}

	public static final int COMMON_ACTOR_COUNT = 10;
	public static final int TEST_ACTOR_COUNT = 25;
	public static final int PRODUCER_ACTOR_COUNT = 25;

	public static void sleeper(int seconds) {
		int millis = seconds * 1000 + -50 + nextInt(100); // a little
															// variation
		// logger.trace("sleep: %dms", millis);
		sleep(millis);
	}

	public static void dumpMessages(List<DefaultMessage> messages) {
		synchronized (messages) {
			if (messages.size() > 0) {
				for (DefaultMessage m : messages) {
					logger.info("%s", m);
				}
			}
		}
	}

	protected List<ChangeListener> listeners = new LinkedList<ChangeListener>();

	public void addChangeListener(ChangeListener l) {
		if (!listeners.contains(l)) {
			listeners.add(l);
		}
	}

	public void removeChangeListener(ChangeListener l) {
		listeners.remove(l);
	}

	protected void fireChangeListeners(ChangeEvent e) {
		for (ChangeListener l : listeners) {
			l.stateChanged(e);
		}
	}

	protected static String[] types = new String[] { "widget", "framit", "frizzle", "gothca", "splat" };

	public static String[] getItemTypes() {
		return types;
	}

	public static void main(String[] args) {
		DefaultActorTest at = new DefaultActorTest();
		at.run(args);
		logger.trace("Done");
	}

	protected String title;

	public String getTitle() {
		return title;
	}

	

	volatile protected boolean done;

	public void terminateRun() {
		done = true;
	}

	public static String[] getTestNames() {
		return new String[] { "Countdown", "Producer Consumer", /* "Quicksort", */"MapReduce", "Virus Scan", "All" };
	}

	DefaultActorManager actorManager;

	public DefaultActorManager getActorManager() {
		return actorManager;
	}

	public void setActorManager(DefaultActorManager actorManager) {
		this.actorManager = actorManager;
	}

	public void run(String[] args) {
		done = false;
		// DefaultLogger.getDefaultInstance().setIncludeDate(false);
		DefaultLogger.getDefaultInstance().setIncludeContext(false);
		DefaultLogger.getDefaultInstance().setIncludeCaller(false);
		// DefaultLogger.getDefaultInstance().setIncludeThread(false);
		DefaultLogger.getDefaultInstance().setLogToFile(false);
		DefaultLogger.getDefaultInstance().setThreadFieldWidth(10);

		int sc = stepCount;
		int tc = threadCount;
		boolean doTest = false;
		title = "";
		
		if (!doTest ) {
			doTest = true;
		}
		if (doTest) {
			if (title.length() > 0) {
				title += " ";
			}
			title += "(Countdown Test)";
		}
		
		DefaultActorManager am = getManager();
		try {
			Map<String, Object> options = new HashMap<String, Object>();
			options.put(DefaultActorManager.ACTOR_THREAD_COUNT, tc);
			am.initialize(options);
			if (doTest) {
				for (int i = 0; i < COMMON_ACTOR_COUNT; i++) {
					Actor a = am.createActor(TestActor.class, String.format("common%02d", i));
					if (a instanceof TestableActor) {
						TestableActor ta = (TestableActor) a;
						ta.setActorTest(this);
					}
					a.setCategory(TestActor.class.getSimpleName());
					getTestActors().put(a.getName(), a);
					// logger.trace("created: %s", a);
				}
				for (int i = 0; i < TEST_ACTOR_COUNT; i++) {
					Actor a = am.createActor(TestActor.class, String.format("actor%02d", i));
					if (a instanceof TestableActor) {
						TestableActor ta = (TestableActor) a;
						ta.setActorTest(this);
					}
					getTestActors().put(a.getName(), a);
					// logger.trace("created: %s", a);
				}
			}

			for (String key : getTestActors().keySet()) {
				am.startActor(getTestActors().get(key));
			}

			for (int i = sc; i > 0; i--) {
				if (done) {
					break;
				}
				// see if idle a while
				long now = new Date().getTime();
				if (am.getActiveRunnableCount() == 0) {
					if (now - am.getLastDispatchTime() > MAX_IDLE_SECONDS * 1000
							&& now - am.getLastSendTime() > MAX_IDLE_SECONDS * 1000) {
						break;
					}
				}
				setStepCount(i);
				fireChangeListeners(new ChangeEvent(this));
				if (i < 10 || i % 10 == 0) {
					logger.trace("main waiting: %d...", i);
				}
				sleeper(1);
			}
			setStepCount(0);
			fireChangeListeners(new ChangeEvent(this));

			// logger.trace("main terminating");
			am.terminateAndWait();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

 

/**
 * An actor that sends messages while counting down a send count. 
 * 
 * @author BFEIGENB
 *
 */
public class TestActor extends TestableActor {

	@Override
	public void activate() {
		logger.trace("TestActor activate: %s", this);
		super.activate();
	}

	@Override
	public void deactivate() {
		logger.trace("TestActor deactivate: %s", this);
		super.deactivate();
	}

	@Override
	protected void runBody() {
		// logger.trace("TestActor:%s runBody: %s", getName(), this);
		DefaultActorTest.sleeper(1);
		DefaultMessage m = new DefaultMessage("init", 8);
		getManager().send(m, null, this);
	}

	@Override
	protected void loopBody(Message m) {
		// logger.trace("TestActor:%s loopBody %s: %s", getName(), m, this);
		DefaultActorTest.sleeper(1);
		String subject = m.getSubject();
		if ("repeat".equals(subject)) {
			int count = (Integer) m.getData();
			logger.trace("TestActor:%s repeat(%d) %s: %s", getName(), count, m,
					this);
			if (count > 0) {
				m = new DefaultMessage("repeat", count - 1);
				// logger.trace("TestActor loopBody send %s: %s", m, this);
				String toName = "actor"
						+ DefaultActorTest
								.nextInt(DefaultActorTest.TEST_ACTOR_COUNT);
				Actor to = actorTest.getTestActors().get(toName);
				if (to != null) {
					getManager().send(m, this, to);
				} else {
					logger.warning("repeat:%s to is null: %s", getName(),
							toName);
				}
			}
		} else if ("init".equals(subject)) {
			int count = (Integer) m.getData();
			count = DefaultActorTest.nextInt(count) + 1;
			logger.trace("TestActor:%s init(%d): %s", getName(), count, this);
			for (int i = 0; i < count; i++) {
				DefaultActorTest.sleeper(1);
				m = new DefaultMessage("repeat", count);
				// logger.trace("TestActor runBody send %s: %s", m, this);
				String toName = "actor"
						+ DefaultActorTest
								.nextInt(DefaultActorTest.TEST_ACTOR_COUNT);
				Actor to = actorTest.getTestActors().get(toName);
				if (to != null) {
					getManager().send(m, this, to);
				} else {
					logger.warning("init:%s to is null: %s", getName(), toName);
				}
				DefaultMessage dm = new DefaultMessage("repeat", count);
				dm.setDelayUntil(new Date().getTime()
						+ (DefaultActorTest.nextInt(5) + 1) * 1000);
				getManager().send(dm, this, this.getClass().getSimpleName());
			}
		} else {
			logger.warning("TestActor:%s loopBody unknown subject: %s",
					getName(), subject);
		}
	}

}

 

 

输出：

 

 

12:59:38.883 T [main      ] - TestActor activate: TestActor[name=common06, category=TestActor, messages=0]
12:59:38.886 T [main      ] - TestActor activate: TestActor[name=common05, category=TestActor, messages=0]
12:59:38.887 T [main      ] - TestActor activate: TestActor[name=common08, category=TestActor, messages=0]
12:59:38.889 T [main      ] - TestActor activate: TestActor[name=common07, category=TestActor, messages=0]
12:59:38.890 T [main      ] - TestActor activate: TestActor[name=common09, category=TestActor, messages=0]
12:59:38.891 T [main      ] - TestActor activate: TestActor[name=common00, category=TestActor, messages=0]
12:59:38.892 T [main      ] - TestActor activate: TestActor[name=common01, category=TestActor, messages=0]
12:59:38.893 T [main      ] - TestActor activate: TestActor[name=common02, category=TestActor, messages=0]
12:59:38.895 T [main      ] - TestActor activate: TestActor[name=common03, category=TestActor, messages=0]
12:59:38.896 T [main      ] - TestActor activate: TestActor[name=common04, category=TestActor, messages=0]
12:59:38.897 T [main      ] - TestActor activate: TestActor[name=actor24, category=default, messages=0]
12:59:38.899 T [main      ] - TestActor activate: TestActor[name=actor11, category=default, messages=0]
12:59:38.904 T [main      ] - TestActor activate: TestActor[name=actor23, category=default, messages=0]
12:59:38.905 T [main      ] - TestActor activate: TestActor[name=actor10, category=default, messages=0]
12:59:38.906 T [main      ] - TestActor activate: TestActor[name=actor22, category=default, messages=0]
12:59:38.907 T [main      ] - TestActor activate: TestActor[name=actor13, category=default, messages=0]
12:59:38.908 T [main      ] - TestActor activate: TestActor[name=actor21, category=default, messages=0]
12:59:38.909 T [main      ] - TestActor activate: TestActor[name=actor12, category=default, messages=0]
12:59:38.910 T [main      ] - TestActor activate: TestActor[name=actor20, category=default, messages=0]
12:59:38.911 T [main      ] - TestActor activate: TestActor[name=actor02, category=default, messages=0]
12:59:38.912 T [main      ] - TestActor activate: TestActor[name=actor01, category=default, messages=0]
12:59:38.914 T [main      ] - TestActor activate: TestActor[name=actor00, category=default, messages=0]
12:59:38.915 T [main      ] - TestActor activate: TestActor[name=actor19, category=default, messages=0]
12:59:38.916 T [main      ] - TestActor activate: TestActor[name=actor06, category=default, messages=0]
12:59:38.917 T [main      ] - TestActor activate: TestActor[name=actor18, category=default, messages=0]
12:59:38.917 T [main      ] - TestActor activate: TestActor[name=actor05, category=default, messages=0]
12:59:38.918 T [main      ] - TestActor activate: TestActor[name=actor04, category=default, messages=0]
12:59:38.919 T [main      ] - TestActor activate: TestActor[name=actor03, category=default, messages=0]
12:59:38.920 T [main      ] - TestActor activate: TestActor[name=actor15, category=default, messages=0]
12:59:38.921 T [main      ] - TestActor activate: TestActor[name=actor14, category=default, messages=0]
12:59:38.922 T [main      ] - TestActor activate: TestActor[name=actor09, category=default, messages=0]
12:59:38.923 T [main      ] - TestActor activate: TestActor[name=actor17, category=default, messages=0]
12:59:38.923 T [main      ] - TestActor activate: TestActor[name=actor08, category=default, messages=0]
12:59:38.924 T [main      ] - TestActor activate: TestActor[name=actor16, category=default, messages=0]
12:59:38.925 T [main      ] - TestActor activate: TestActor[name=actor07, category=default, messages=0]
12:59:38.926 T [main      ] - main waiting: 120...
12:59:42.970 T [actor3    ] - TestActor:common06 init(4): TestActor[name=common06, category=TestActor, messages=0]
12:59:43.028 T [actor2    ] - TestActor:common03 init(8): TestActor[name=common03, category=TestActor, messages=0]
12:59:43.048 T [actor1    ] - TestActor:common01 init(6): TestActor[name=common01, category=TestActor, messages=0]
12:59:43.054 T [actor8    ] - TestActor:common05 init(4): TestActor[name=common05, category=TestActor, messages=0]
12:59:43.064 T [actor6    ] - TestActor:common09 init(8): TestActor[name=common09, category=TestActor, messages=0]
12:59:43.873 T [actor0    ] - TestActor:common02 init(7): TestActor[name=common02, category=TestActor, messages=0]
12:59:43.898 T [actor9    ] - TestActor:common08 init(1): TestActor[name=common08, category=TestActor, messages=0]
12:59:43.993 T [actor7    ] - TestActor:common04 init(2): TestActor[name=common04, category=TestActor, messages=1]
12:59:43.994 T [actor5    ] - TestActor:common00 init(1): TestActor[name=common00, category=TestActor, messages=0]
12:59:43.996 W [actor2    ] - init:common03 to is null: actor5
12:59:44.039 W [actor8    ] - init:common05 to is null: actor9
12:59:44.052 W [actor1    ] - init:common01 to is null: actor9
12:59:44.060 T [actor4    ] - TestActor:common07 init(5): TestActor[name=common07, category=TestActor, messages=0]
12:59:44.912 W [actor0    ] - init:common02 to is null: actor6
12:59:44.968 W [actor5    ] - init:common00 to is null: actor9
12:59:44.986 W [actor3    ] - init:common06 to is null: actor1
12:59:44.986 W [actor7    ] - init:common04 to is null: actor9
12:59:45.108 W [actor4    ] - init:common07 to is null: actor8
12:59:45.947 T [actor9    ] - TestActor:actor13 init(7): TestActor[name=actor13, category=default, messages=0]
12:59:45.951 T [actor5    ] - TestActor:actor22 init(1): TestActor[name=actor22, category=default, messages=1]
12:59:45.976 W [actor3    ] - init:common06 to is null: actor1
12:59:46.016 W [actor2    ] - init:common03 to is null: actor6
12:59:46.052 W [actor1    ] - init:common01 to is null: actor6
12:59:46.115 W [actor4    ] - init:common07 to is null: actor9
12:59:46.932 W [actor3    ] - init:common06 to is null: actor5
12:59:46.940 W [actor5    ] - init:actor22 to is null: actor3
12:59:47.039 T [actor7    ] - TestActor:actor24 init(8): TestActor[name=actor24, category=default, messages=1]
12:59:47.042 W [actor1    ] - init:common01 to is null: actor6
12:59:47.059 W [actor2    ] - init:common03 to is null: actor0
12:59:47.904 T [actor8    ] - TestActor:actor21 init(7): TestActor[name=actor21, category=default, messages=1]
12:59:47.964 T [actor3    ] - TestActor:actor10 init(4): TestActor[name=actor10, category=default, messages=0]
12:59:47.988 T [actor5    ] - TestActor:actor23 init(7): TestActor[name=actor23, category=default, messages=2]
12:59:48.003 W [actor6    ] - init:common09 to is null: actor3
12:59:48.042 W [actor1    ] - init:common01 to is null: actor6
12:59:48.072 W [actor2    ] - init:common03 to is null: actor8
12:59:48.101 W [actor4    ] - init:common07 to is null: actor6
12:59:48.880 W [actor8    ] - init:actor21 to is null: actor8
12:59:48.903 W [actor9    ] - init:actor13 to is null: actor3
12:59:48.952 T [main      ] - main waiting: 110...
12:59:49.024 W [actor2    ] - init:common03 to is null: actor3
12:59:49.037 W [actor6    ] - init:common09 to is null: actor9
12:59:49.085 W [actor4    ] - init:common07 to is null: actor6

 

源代码打包在附件中。

评论

1 楼 m635674608 2015-03-29  

这样看起来，，和线程是一样的。性能也是一样的。。？？