并发设计模式之Guarded Suspension模式

Posted 2020-10-22 zzfx

- 原文链接： http://www.joyhwong.com/2016/11/19/并发设计模式之guarded-suspension模式/

Guarded Suspension意为保护暂停，其核心思想是仅当服务进程准备好时，才提供服务。设想一种场景，服务器可能会在很短时间内承受大量的客户端请求，客户端请求的数量可能超过服务器本身的即时处理能力，而服务端程序又不能丢弃任何一个客户请求。此时，最佳的处理方案莫过于让客户端要求进行排队，由服务端程序一个接一个处理。这样，既保证了所有的客户端请求均不丢失，同时也避免了服务器由于同时处理太多的请求而崩溃

1.Guarded Suspension模式的结构

Guarded Suspension模式的主要成员有：Request、RequestQueue、ClientThread、 ServerThread

• Request：表示客户端请求
• RequestQueue：用于保存客户端请求队列
• ClientThread：客户端进程
• ServerThread：服务器进程

其中，ClientThread负责不断发起请求，并将请求对象放入请求队列。ServerThread则根据其自身的状态，在有能力处理请求时，从RequestQueue中提取请求对象加以处理。系统的工作流程如图

从流程图中可以看到，客户端的请求数量超过了服务线程的能力。在频繁的客户端请求中，RequestQueue充当了中间缓存，存放未处理的请求，保证了客户请求不丢失，同时也保护了服务线程不会受到大量并发的请求，而导致计算机资源不足

2. Guarded Suspension模式的简单实现

Request类封装了请求的内容

package com.joyhwong;

public class Request {
	private String name;
	
	public Request(String name) {
		this.name = name;
	}
	
	public String getName() {
		return name;
	}
	
	@Override
	public String toString() {
		return "[ Request "+ name + " ]"; 
	}
}

RequestQueue对象作为Request的集合，维护系统的Request请求列表

package com.joyhwong;

import java.util.LinkedList;

public class RequestQueue {
	private LinkedList<Request> queue = new LinkedList<>();
	public synchronized Request getRequest() {
		while (queue.size() == 0) {
			try {
				wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return queue.removeFirst();
	}
	
	public synchronized void addRequest(Request request) {
		queue.add(request);
		notifyAll();
	}
}

服务端进程用于处理用户的请求操作，代码实现如下

package com.joyhwong;

public class ServerThread extends Thread {
	private RequestQueue requestQueue;
	
	public ServerThread(RequestQueue requestQueue, String name) {
		super(name);
		this.requestQueue = requestQueue;
	}
	
	@Override
	public void run() {
		while (true) {
			final Request request = requestQueue.getRequest();
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(Thread.currentThread().getName() + " handles " + request);
		}
	}
}

客户端的请求发起进程代码实现如下

package com.joyhwong;

public class ClientThread extends Thread {
	private RequestQueue requestQueue;
	
	public ClientThread(RequestQueue requestQueue, String name) {
		super(name);
		this.requestQueue = requestQueue;
	}
	
	@Override
	public void run() {
		for (int i = 0; i < 10; i++) {
			Request request = new Request("RequestID: " + i + " Thread_Name: " + Thread.currentThread().getName());
			System.out.println(Thread.currentThread().getName() + " requests " + request);
			requestQueue.addRequest(request);
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println("ClientThread Name is: " + Thread.currentThread().getName());
		}
		System.out.println(Thread.currentThread().getName() + "request end");
	}
}

主函数如下

package com.joyhwong;

public class Main {
	public static void main(String[] args) {
		RequestQueue requestQueue = new RequestQueue();
		for (int i = 0; i < 10; i++) {
			new ServerThread(requestQueue, "ServerThread" + i).start();
		}
		
		for (int i = 0; i < 10; i++) {
			new ClientThread(requestQueue, "ClientThread" + i).start();
		}
	}
}