clojure中的agent实现

最近看了下clojure的并发,其中提到了agent.agent的原理是把对agent的action提交到线程池中运行.它保证线程安全的措施是对每一个agent,同时只有一个actio
在运行.

下面做一个简单的对比,多个线程多次对同一个StringBuilder进行append,如果不同步,结果是不正确的.

		final StringBuilder sb = new StringBuilder();
		ExecutorService pool = Executors.newFixedThreadPool(Runtime
				.getRuntime().availableProcessors() + 2);
		for (int i = 0; i < THREAD_NUMBER; i++) {
			pool.submit(new Runnable() {
				@Override
				public void run() {
					for (int j = 0; j < ITERATION_NUMBER; j++) {
						sb.append(bar);
					}
				}

			});
		}
		pool.shutdown();
		pool.awaitTermination(60, TimeUnit.SECONDS);
		String s = sb.toString();
		System.out.println(s.length() / bar.length());

输出结果不是ITERATION_NUMBER * THREAD_NUMBER

可以用synchronized或者java 1.5里的lock来同步,也可以仿照agent的原理,不使用同步机制,而是让action顺序执行

结果表明,让action顺序执行,而不使用lock或者synchronized等手段,效率确实比较高.

package foo.concurrency;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ClojureAgent {

	private static String bar = "log from full line a very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very very long line\n";

	private static int ITERATION_NUMBER = 4000;

	private static int THREAD_NUMBER = 6;

	public static void main(String[] args) throws Exception {
		long start = System.currentTimeMillis();
		base();
		System.out.println("time is " + (System.currentTimeMillis() - start));
		start = System.currentTimeMillis();
		sync();
		System.out.println("time is " + (System.currentTimeMillis() - start));
		start = System.currentTimeMillis();
		lock();
		System.out.println("time is " + (System.currentTimeMillis() - start));
		start = System.currentTimeMillis();
		pool();
		System.out.println("time is " + (System.currentTimeMillis() - start));
	}

	private static void pool() throws Exception {

		final StringBuilder sb = new StringBuilder();
		ExecutorService pool = Executors.newFixedThreadPool(Runtime
				.getRuntime().availableProcessors() + 2);
		for (int i = 0; i < THREAD_NUMBER; i++) {
			pool.submit(new Callable<Object>() {

				@Override
				public Object call() throws Exception {
					for (int j = 0; j < ITERATION_NUMBER; j++) {
						sb.append(bar);
					}
					return null;
				}

			}).get();
		}
		pool.shutdown();
		pool.awaitTermination(60, TimeUnit.SECONDS);
		String s = sb.toString();
		System.out.println(s.length() / bar.length());
	}

	private static void sync() throws Exception {

		final StringBuilder sb = new StringBuilder();
		ExecutorService pool = Executors.newFixedThreadPool(Runtime
				.getRuntime().availableProcessors() + 2);
		for (int i = 0; i < THREAD_NUMBER; i++) {
			pool.submit(new Runnable() {
				@Override
				public void run() {

					for (int j = 0; j < ITERATION_NUMBER; j++) {
						synchronized (sb) {
							sb.append(bar);
						}
					}

				}

			});
		}
		pool.shutdown();
		pool.awaitTermination(60, TimeUnit.SECONDS);
		String s = sb.toString();
		System.out.println(s.length() / bar.length());
	}

	private static void lock() throws Exception {
		final Lock lock = new ReentrantLock();
		final StringBuilder sb = new StringBuilder();
		ExecutorService pool = Executors.newFixedThreadPool(Runtime
				.getRuntime().availableProcessors() + 2);
		for (int i = 0; i < THREAD_NUMBER; i++) {
			pool.submit(new Runnable() {
				@Override
				public void run() {

					for (int j = 0; j < ITERATION_NUMBER; j++) {
						lock.lock();
						try {
							sb.append(bar);
						} finally {
							lock.unlock();
						}
					}
				}

			});
		}
		pool.shutdown();
		pool.awaitTermination(60, TimeUnit.SECONDS);
		String s = sb.toString();
		System.out.println(s.length() / bar.length());
	}

	private static void base() throws Exception {

		final StringBuilder sb = new StringBuilder();
		ExecutorService pool = Executors.newFixedThreadPool(Runtime
				.getRuntime().availableProcessors() + 2);
		for (int i = 0; i < THREAD_NUMBER; i++) {
			pool.submit(new Runnable() {
				@Override
				public void run() {
					for (int j = 0; j < ITERATION_NUMBER; j++) {
						sb.append(bar);
					}
				}

			});
		}
		pool.shutdown();
		pool.awaitTermination(60, TimeUnit.SECONDS);
		String s = sb.toString();
		System.out.println(s.length() / bar.length());
	}
}