Multithreading mode (I) -- protective pause and join source code

Definition

namely Guarded Suspension, It is used when one thread waits for the execution result of another thread

The main points of ：

• There is a result that needs to be passed from one thread to another , Let them relate to the same GuardedObject（ Bridge as result transmission ）
• If there are results continuously from one thread to another, you can use message queuing （ See the producer / consumer ）
• JDK in ,join The implementation of the ,Future The implementation of the , That's the model
• Because we have to wait for the outcome of the other side , So it's classified as synchronous mode

Realization

Code example ：

public class Test1 {
    
    //  Threads 1  Wait for thread 2 Download Results for 
    public static void main(String[] args) {
    
        GuardedObject guardedObject = new GuardedObject();
        new Thread(() -> {
    
            //  Wait for the result 
            log.debug(" Wait for the result ");
            try {
    
                Object a = guardedObject.get();
                log.debug(" The result is ：{}", a);
            } catch (InterruptedException e) {
    
                e.printStackTrace();
            }
        }, "t1").start();
        new Thread(() -> {
    
            //  Perform the operation 
            log.debug(" To get the results ");
            guardedObject.complete(1);
        }, "t2").start();
    }
}

class GuardedObject {
    
    //  result 
    private Object response;

    //  To get the results 
    public Object get() throws InterruptedException {
    
        synchronized (this) {
    
            //  No results 
            while (response == null) {
    
                this.wait();
            }
            return response;
        }
    }

    //  Produce results 
    public void complete(Object response) {
    
        synchronized (this) {
    
            //  Assign a value to a member variable 
            this.response = response;
            this.notifyAll();
        }
    }
}

result ： It can be seen that the results can be 2 Thread sharing

17:20:14.141 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  Wait for the result 
17:20:14.141 [t2] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  To get the results 
17:20:14.145 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  The result is ：1

Expand 1： Increase timeout

example 1： As a result, it was delivered on time

@Slf4j()
public class Test1 {
    
    //  Threads 1  Wait for thread 2 Download Results for 
    public static void main(String[] args) {
    
        GuardedObject guardedObject = new GuardedObject();
        new Thread(() -> {
    
            //  Wait for the result 
            log.debug(" Wait for the result ");
            try {
    
                Object a = guardedObject.get(2000);
                log.debug(" The result is ：{}", a);
            } catch (InterruptedException e) {
    
                e.printStackTrace();
            }
        }, "t1").start();
        new Thread(() -> {
    
            //  Perform the operation 
            log.debug(" To get the results ");
            try {
    
                Thread.sleep(1000);
            } catch (InterruptedException e) {
    
                e.printStackTrace();
            }
            guardedObject.complete(1);
        }, "t2").start();
    }
}

class GuardedObject {
    
    //  result 
    private Object response;

    //  To get the results 
    //  Expand 1：timeout Indicates how long to wait 
    public Object get(long timeout) throws InterruptedException {
    
        synchronized (this) {
    
            //  Starting time 
            long start = System.currentTimeMillis();
            //  Experience time 
            long passTime = 0;
            //  No results 
            while (response == null) {
    
                //  When the elapsed time exceeds the maximum waiting time , Exit loop 
                if (passTime >= timeout) {
    
                    break;
                }
                //this.wait(n); //  No results have been obtained yet , Awakened by falsehood 
                this.wait(timeout - passTime);
                //  Find the experience time 
                passTime = System.currentTimeMillis() - start;
            }
            return response;
        }
    }

    //  Produce results 
    public void complete(Object response) {
    
        synchronized (this) {
    
            //  Assign a value to a member variable 
            this.response = response;
            this.notifyAll();
        }
    }
}

result ：

17:36:53.778 [t2] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  To get the results 
17:36:53.778 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  Wait for the result 
17:36:54.802 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  The result is ：1

example 2： Sleep the above 1 Seconds to sleep 3 second

result ：

17:38:27.898 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  Wait for the result 
17:38:27.898 [t2] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  To get the results 
17:38:29.916 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  The result is ：null

example 3： spurious wakeup

Put the above complete(1) Change to complete(null)：
result ：

17:41:09.282 [t2] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  To get the results 
17:41:09.282 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  Wait for the result 
17:41:11.298 [t1] DEBUG com.yhx.toali.multiThread.GuardedObject.Test1 -  The result is ：null

join Source check ：

public final synchronized void join(final long millis) throws InterruptedException {
    
    if (millis > 0) {
    
        if (isAlive()) {
    
            final long startTime = System.nanoTime();
            long delay = millis;
            do {
    
                wait(delay);
            } while (isAlive() && (delay = millis -
                    TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime)) > 0);
        }
    } else if (millis == 0) {
    
        while (isAlive()) {
    
            wait(0);
        }
    } else {
    
        throw new IllegalArgumentException("timeout value is negative");
    }
}

Expand 2： Result waiting and production

Multitask version GuardedObject In the figure Futures It's like a mailbox on the first floor of a residential building （ Each mailbox has a room number ）, Left side t0,t2,t4 It's like residents waiting for mail , On the right side of the t1,t3,t5 Like a postman

If you need to use between multiple classes GuardedObject object , Passing as a parameter is not very convenient , Therefore, an intermediate class for decoupling is designed , This not only decouples 【 As a result, the waiting person 】 and 【 As a result, the producer 】, It can also support the management of multiple tasks at the same time .

The difference from the producer consumer model is ： There is a one-to-one correspondence between producers and consumers , But the producer consumer model is not .rpc This pattern is used in the invocation of the framework .

example ：

public class Test1 {
    
    //  Threads 1  Wait for thread 2 Download Results for 
    public static void main(String[] args) throws InterruptedException {
    
        for (int i = 0; i < 3; i++) {
    
            new People().start();
        }
        Thread.sleep(1000);
        for (Integer id : MailBoxes.getIds()) {
    
            new Postman(id, " Content " + id).start();
        }
    }
}

@Slf4j(topic = "people")
class People extends Thread{
    
    @SneakyThrows
    @Override
    public void run() {
    
        // Receive letters 
        GuardedObject guardedObject = MailBoxes.createGuardedObject();
        log.debug(" Receive letters ,id:{}", guardedObject.getId());
        Object mail = guardedObject.get(5000);
        log.debug(" Received a letter ,id:{}, Content :{}", guardedObject.getId(), mail);
    }
}

@Slf4j(topic = "postman")
class Postman extends Thread{
    
    private int id;
    private String mail;

    public Postman(int id, String mail) {
    
        this.id = id;
        this.mail = mail;
    }

    @Override
    public void run() {
    
        GuardedObject go = MailBoxes.getGO(id);
        log.debug(" Deliver a letter ,id:{}, Content :{}", id, mail);
        go.complete(mail);
    }
}

//  Decoupling class （ generic class ）, mailbox 
class MailBoxes {
    
    private static Map<Integer, GuardedObject> boxes = new Hashtable<>();
    private static int id = 1;
    //  Produce a unique id
    private static synchronized int generateId() {
    
        return id++;
    }

    public static GuardedObject getGO(int id) {
    
        return boxes.remove(id);
    }

    public static GuardedObject createGuardedObject() {
    
        GuardedObject go = new GuardedObject(generateId());
        boxes.put(go.getId(), go);
        return go;
    }

    public static Set<Integer> getIds() {
    
        return boxes.keySet();
    }
}

class GuardedObject {
    
    // id identification  guardedObject
    private int id;

    public GuardedObject(int generateId) {
    
        this.id = generateId;
    }

    public int getId() {
    
        return id;
    }

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

    //  result 
    private Object response;

    //  To get the results 
    //  Expand 1：timeout Indicates how long to wait 
    public Object get(long timeout) throws InterruptedException {
    
        synchronized (this) {
    
            //  Starting time 
            long start = System.currentTimeMillis();
            //  Experience time 
            long passTime = 0;
            //  No results 
            while (response == null) {
    
                //  When the elapsed time exceeds the maximum waiting time , Exit loop 
                if (passTime >= timeout) {
    
                    break;
                }
                //this.wait(n); //  No results have been obtained yet , Awakened by falsehood 
                this.wait(timeout - passTime);
                //  Find the experience time 
                passTime = System.currentTimeMillis() - start;
            }
            return response;
        }
    }

    //  Produce results 
    public void complete(Object response) {
    
        synchronized (this) {
    
            //  Assign a value to a member variable 
            this.response = response;
            this.notifyAll();
        }
    }
}