|
Future模式
先返回空殼數據，同時后臺啟動線程執行任務，當獲取數據時，使用wait方法進行阻塞，等待任務執行完畢，再進行返回真實數據
入口
public class Main {
public static void main(String[] args) throws InterruptedException {
FutureClient fc = new FutureClient();
Data data = fc.request("請求參數");
System.out.println("請求發送成功!");
System.out.println("做其他的事情...");
String result = data.getRequest();
System.out.println(result);
}
}
對請求進行假數據響應
public class FutureClient {
public Data request(final String queryStr){
//1 我想要一個代理對象（Data接口的實現類）先返回給發送請求的客戶端，告訴他請求已經接收到，可以做其他的事情
final FutureData futureData = new FutureData();
//2 啟動一個新的線程，去加載真實的數據，傳遞給這個代理對象
new Thread(new Runnable() {
@Override
public void run() {
//3 這個新的線程可以去慢慢的加載真實對象，然后傳遞給代理對象
RealData realData = new RealData(queryStr);
futureData.setRealData(realData);
}
}).start();
return futureData;
}
}
數據抽象
public interface Data {
String getRequest();
}
假數據
public class FutureData implements Data{
private RealData realData ;
private boolean isReady = false;
public synchronized void setRealData(RealData realData) {
//如果已經裝載完畢了，就直接返回
if(isReady){
return;
}
//如果沒裝載，進行裝載真實對象
this.realData = realData;
isReady = true;
//進行通知
notify();
}
@Override
public synchronized String getRequest() {
//如果沒裝載好 程序就一直處于阻塞狀態
while(!isReady){
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//裝載好直接獲取數據即可
return this.realData.getRequest();
}
}
真實數據
public class RealData implements Data{
private String result ;
public RealData (String queryStr){
System.out.println("根據" + queryStr + "進行查詢，這是一個很耗時的操作..");
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("操作完畢，獲取結果");
result = "查詢結果";
}
@Override
public String getRequest() {
return result;
}
}
- 最常用的并行計算模式，系統由倆類進程協作進行工作，Master負責接收和分配任務，Worker負責處理子任務，當各個worker子進程處理完之后，會將結果返回給Master，由Master進行歸納和總結，將任務分解成若干個子任務，并行執行，從而提高系統的吞吐量
入口
import java.util.Random;
public class Main {
public static void main(String[] args) {
Master master = new Master(new Worker(), 20);
Random r = new Random();
for(int i = 1; i <= 100; i++){
Task t = new Task();
t.setId(i);
t.setPrice(r.nextInt(1000));
master.submit(t);
}
master.execute();
long start = System.currentTimeMillis();
while(true){
if(master.isComplete()){
long end = System.currentTimeMillis() - start;
int priceResult = master.getResult();
System.out.println("最終結果：" + priceResult + ", 執行時間：" + end);
break;
}
}
}
}
任務
public class Task {
private int id;
private int price ;
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public int getPrice() {
return price;
}
public void setPrice(int price) {
this.price = price;
}
}
Master
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
public class Master {
//1 有一個盛放任務的容器
private ConcurrentLinkedQueue workQueue = new ConcurrentLinkedQueue();
//2 需要有一個盛放worker的集合
private HashMap workers = new HashMap();
//3 需要有一個盛放每一個worker執行任務的結果集合 workers可能并發的提交結果集
private ConcurrentHashMap resultMap = new ConcurrentHashMap();
//4 構造方法
public Master(Worker worker , int workerCount){
worker.setWorkQueue(this.workQueue);
worker.setResultMap(this.resultMap);
for(int i = 0; i < workerCount; i ++){
this.workers.put(Integer.toString(i), new Thread(worker));
}
}
//5 需要一個提交任務的方法
public void submit(Task task){
this.workQueue.add(task);
}
//6 需要有一個執行的方法，啟動所有的worker方法去執行任務
public void execute(){
for(Map.Entry me : workers.entrySet()){
me.getValue().start();
}
}
//7 判斷是否運行結束的方法
public boolean isComplete() {
for(Map.Entry me : workers.entrySet()){
if(me.getValue().getState() != Thread.State.TERMINATED){
return false;
}
}
return true;
}
//8 計算結果方法
public int getResult() {
int priceResult = 0;
for(Map.Entry me : resultMap.entrySet()){
priceResult += (Integer)me.getValue();
}
return priceResult;
}
Worker
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
public class Worker implements Runnable {
private ConcurrentLinkedQueue workQueue;
private ConcurrentHashMap resultMap;
public void setWorkQueue(ConcurrentLinkedQueue workQueue) {
this.workQueue = workQueue;
}
public void setResultMap(ConcurrentHashMap resultMap) {
this.resultMap = resultMap;
}
@Override
public void run() {
while(true){
Task input = this.workQueue.poll();
if(input == null) break;
Object output = handle(input);
this.resultMap.put(Integer.toString(input.getId()), output);
}
}
private Object handle(Task input) {
Object output = null;
try {
//處理任務的耗時。。 比如說進行操作數據庫。。。
Thread.sleep(500);
output = input.getPrice();
} catch (InterruptedException e) {
e.printStackTrace();
}
return output;
}
}
入口
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class Main {
public static void main(String[] args) throws Exception {
//內存緩沖區
BlockingQueue queue = new LinkedBlockingQueue(10);
//生產者
Provider p1 = new Provider(queue);
Provider p2 = new Provider(queue);
Provider p3 = new Provider(queue);
//消費者
Consumer c1 = new Consumer(queue);
Consumer c2 = new Consumer(queue);
Consumer c3 = new Consumer(queue);
//創建線程池運行,這是一個緩存的線程池，可以創建無窮大的線程，沒有任務的時候不創建線程。空閑線程存活時間為60s（默認值）
ExecutorService cachePool = Executors.newCachedThreadPool();
cachePool.execute(p1);
cachePool.execute(p2);
cachePool.execute(p3);
cachePool.execute(c1);
cachePool.execute(c2);
cachePool.execute(c3);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
p1.stop();
p2.stop();
p3.stop();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// cachePool.shutdown();
// cachePool.shutdownNow();
}
}
數據模型
public final class Data {
private String id;
private String name;
public Data(String id, String name){
this.id = id;
this.name = name;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString(){
return "{id: " + id + ", name: " + name + "}";
}
}
生產者
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Provider implements Runnable{
//共享緩存區
private BlockingQueue queue;
//多線程間是否啟動變量，有強制從主內存中刷新的功能。即時返回線程的狀態
private volatile boolean isRunning = true;
//id生成器
private static AtomicInteger count = new AtomicInteger();
//隨機對象
private static Random r = new Random();
public Provider(BlockingQueue queue){
this.queue = queue;
}
@Override
public void run() {
while(isRunning){
try {
//隨機休眠0 - 1000 毫秒 表示獲取數據(產生數據的耗時)
Thread.sleep(r.nextInt(1000));
//獲取的數據進行累計...
int id = count.incrementAndGet();
//比如通過一個getData方法獲取了
Data data = new Data(Integer.toString(id), "數據" + id);
System.out.println("當前線程:" + Thread.currentThread().getName() + ", 獲取了數據，id為:" + id + ", 進行裝載到公共緩沖區中...");
if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
System.out.println("提交緩沖區數據失敗....");
//do something... 比如重新提交
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void stop(){
this.isRunning = false;
}
}
消費者
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public class Consumer implements Runnable{
private BlockingQueue queue;
public Consumer(BlockingQueue queue){
this.queue = queue;
}
//隨機對象
private static Random r = new Random();
@Override
public void run() {
while(true){
try {
//獲取數據
Data data = this.queue.take();
//進行數據處理。休眠0 - 1000毫秒模擬耗時
Thread.sleep(r.nextInt(1000));
System.out.println("當前消費線程：" + Thread.currentThread().getName() + "， 消費成功，消費數據為id: " + data.getId());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
|
