29、如何使用设计模式优化并发编程

public class ContextTest {

	// 上下文类
	public class Context {
		private String name;
		private long id

		public long getId() {
			return id;
		}

		public void setId(long id) {
			this.id = id;
		}

		public String getName() {
			return this.name;
		}

		public void setName(String name) {
			this.name = name;
		}
	}

	// 设置上下文名字
	public class QueryNameAction {
		public void execute(Context context) {
			try {
				Thread.sleep(1000L);
				String name = Thread.currentThread().getName();
				context.setName(name);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	// 设置上下文ID
	public class QueryIdAction {
		public void execute(Context context) {
			try {
				Thread.sleep(1000L);
				long id = Thread.currentThread().getId();
				context.setId(id);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	// 执行方法
	public class ExecutionTask implements Runnable {

		private QueryNameAction queryNameAction = new QueryNameAction();
		private QueryIdAction queryIdAction = new QueryIdAction();

		@Override
		public void run() {
			final Context context = new Context();
			queryNameAction.execute(context);
			System.out.println("The name query successful");
			queryIdAction.execute(context);
			System.out.println("The id query successful");

			System.out.println("The Name is " + context.getName() + " and id " + context.getId());
		}
	}

	public static void main(String[] args) {
		IntStream.range(1, 5).forEach(i -> new Thread(new ContextTest().new ExecutionTask()).start());
	}
}

The name query successful
The name query successful
The name query successful
The name query successful
The id query successful
The id query successful
The id query successful
The id query successful
The Name is Thread-1 and id 11
The Name is Thread-2 and id 12
The Name is Thread-3 and id 13
The Name is Thread-0 and id 10

private void testThreadLocal() {
    Thread t = new Thread() {
        ThreadLocal<String> mStringThreadLocal = new ThreadLocal<String>();

        @Override
        public void run() {
            super.run();
            mStringThreadLocal.set("test");
            mStringThreadLocal.get();
        }
    };

    t.start();
}

public class ContextTest {
	// 上下文类
	public static class Context {
		private String name;
		private long id;

		public long getId() {
			return id;
		}

		public void setId(long id) {
			this.id = id;
		}

		public String getName() {
			return this.name;
		}

		public void setName(String name) {
			this.name = name;
		}
	}

	// 复制上下文到ThreadLocal中
	public final static class ActionContext {

		private static final ThreadLocal<Context> threadLocal = new ThreadLocal<Context>() {
			@Override
			protected Context initialValue() {
				return new Context();
			}
		};

		public static ActionContext getActionContext() {
			return ContextHolder.actionContext;
		}

		public Context getContext() {
			return threadLocal.get();
		}

		// 获取ActionContext单例
		public static class ContextHolder {
			private final static ActionContext actionContext = new ActionContext();
		}
	}

	// 设置上下文名字
	public class QueryNameAction {
		public void execute() {
			try {
				Thread.sleep(1000L);
				String name = Thread.currentThread().getName();
				ActionContext.getActionContext().getContext().setName(name);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	// 设置上下文ID
	public class QueryIdAction {
		public void execute() {
			try {
				Thread.sleep(1000L);
				long id = Thread.currentThread().getId();
				ActionContext.getActionContext().getContext().setId(id);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	// 执行方法
	public class ExecutionTask implements Runnable {
		private QueryNameAction queryNameAction = new QueryNameAction();
		private QueryIdAction queryIdAction = new QueryIdAction();

		@Override
		public void run() {
			queryNameAction.execute();//设置线程名
			System.out.println("The name query successful");
			queryIdAction.execute();//设置线程ID
			System.out.println("The id query successful");

			System.out.println("The Name is " + ActionContext.getActionContext().getContext().getName() + " and id " + ActionContext.getActionContext().getContext().getId())
		}
	}

	public static void main(String[] args) {
		IntStream.range(1, 5).forEach(i -> new Thread(new ContextTest().new ExecutionTask()).start());
	}
}

The name query successful
The name query successful
The name query successful
The name query successful
The id query successful
The id query successful
The id query successful
The id query successful
The Name is Thread-2 and id 12
The Name is Thread-0 and id 10
The Name is Thread-1 and id 11
The Name is Thread-3 and id 13

//IO处理
public class ServerHandler implements Runnable{
	private Socket socket;
	 
    public ServerHandler(Socket socket) {
        this.socket = socket;
    }
    
    public void run() {
        BufferedReader in = null;
        PrintWriter out = null;
        String msg = null;
        try {
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream(),true);
            while ((msg = in.readLine()) != null && msg.length()!=0) {//当连接成功后在此等待接收消息（挂起，进入阻塞状态）
                System.out.println("server received : " + msg);
                out.print("received~\n");
                out.flush();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        	try {
                in.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                out.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

//Socket启动服务
public class Server {
	
	private static int DEFAULT_PORT = 12345;
	private static ServerSocket server;

	public static void start() throws IOException {
		start(DEFAULT_PORT);
	}

	public static void start(int port) throws IOException {
		if (server != null) {
			return;
		}
		
		try {
			//启动服务
			server = new ServerSocket(port);
			// 通过无线循环监听客户端连接
			while (true) {
				
				Socket socket = server.accept();
				// 当有新的客户端接入时，会执行下面的代码
				long start = System.currentTimeMillis();
				new Thread(new ServerHandler(socket)).start();
				
				long end = System.currentTimeMillis();

				System.out.println("Spend time is " + (end - start));
			}
		} finally {
			if (server != null) {
				System.out.println("服务器已关闭。");
				server.close();
			}

		}

	}
	
	public static void main(String[] args) throws InterruptedException{

		// 运行服务端
		new Thread(new Runnable() {
			public void run() {
				try {
					Server.start();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}).start();

	}
}

//包裹类
public class Package {
	private String name;
	private String address;

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getAddress() {
		return address;
	}

	public void setAddress(String address) {
		this.address = address;
	}

	public void execute() {
		System.out.println(Thread.currentThread().getName()+" executed "+this);
	}
}

//流水线
public class PackageChannel {
	private final static int MAX_PACKAGE_NUM = 100;

	private final Package[] packageQueue;
	private final Worker[] workerPool;
	private int head;
	private int tail;
	private int count;

	public PackageChannel(int workers) {
		this.packageQueue = new Package[MAX_PACKAGE_NUM];
		this.head = 0;
		this.tail = 0;
		this.count = 0;
		this.workerPool = new Worker[workers];
		this.init();
	}

	private void init() {
		for (int i = 0; i < workerPool.length; i++) {
			workerPool[i] = new Worker("Worker-" + i, this);
		}
	}

	/**
	 * push switch to start all of worker to work
	 */
	public void startWorker() {
		Arrays.asList(workerPool).forEach(Worker::start);
	}

	public synchronized void put(Package packagereq) {
		while (count >= packageQueue.length) {
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		this.packageQueue[tail] = packagereq;
		this.tail = (tail + 1) % packageQueue.length;
		this.count++;
		this.notifyAll();
	}

	public synchronized Package take() {
		while (count <= 0) {
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		Package request = this.packageQueue[head];
		this.head = (this.head + 1) % this.packageQueue.length;
		this.count--;
		this.notifyAll();
		return request;
	}

}

//机器人
public class Worker extends Thread{
	 private static final Random random = new Random(System.currentTimeMillis());
	 private final PackageChannel channel;

	    public Worker(String name, PackageChannel channel) {
	        super(name);
	        this.channel = channel;
	    }

	    @Override
	    public void run() {
	        while (true) {
	            channel.take().execute();

	            try {
	                Thread.sleep(random.nextInt(1000));
	            } catch (InterruptedException e) {
	                e.printStackTrace();
	            }
	        }
	    }

}

public class Test {
	   public static void main(String[] args) {
            //新建8个工人
	        final PackageChannel channel = new PackageChannel(8);
            //开始工作
	        channel.startWorker();
            //为流水线添加包裹
	        for(int i=0; i<100; i++) {
	        	 Package packagereq = new Package();
	 	        packagereq.setAddress("test");
	 	        packagereq.setName("test");
	 	        channel.put(packagereq);
	        }
	    }
}