java - 如何在创建类时启动线程并(重新)使用它多次执行方法

Question

我正在尝试解决一个练习，同时为下一次考试学习。这就是问题

Use Java to implement a MsgBatcher class that holds a (finite) set of messages and sends them, on request, in batch. Suppose we have a Message class with a method void send(). MsgBatcher provides a method void enqueue(Message) to add a new message to the batch. It suspends the caller if the MsgBatcher is full (the maximum number of messages that can be enqueued is provided in the MsgBatcher constructor). A method void sendAll() is also provided to send all messages enqueued up to that moment (it empties the MsgBatcher). Organize synchronization that will take care of the fact that sending a message may take a long time. Optional: implement the sendAll method so that the sending is performed asynchronously w.r.t. the caller (i.e., in a separate thread, which should be started at MsgBatcher creation time and reused for each sending).

到目前为止我已经写了这个

import java.util.ArrayList;

public class MsgBatcher {

    public ArrayList<Message> batch = new ArrayList<Message>();
    int maxSpaces;

    public void MsgBatcher(int max){
        this.maxSpaces= max;
    }
    public synchronized void enqueue(Message m) throws InterruptedException{
        while(batch.size() == maxSpaces) wait();
        this.batch.add(m);
    }
    public void sendAll(){
        ArrayList<Message> toSend = new ArrayList<Message>();
        toSend.addAll(batch);
        batch.clear(); //in this way i can accept other messages while sending the others

        for (Message m : toSend){
            m.send();
        }
        batch.clear();
    }
}   

由于 Message 类并不是很重要，所以我只编写了几行来模拟所要求的行为

public class Message {

 public void send(){
     System.out.println("Sending");
     try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
     System.out.println("Sent");
 }
}

我不明白的是如何处理

Organize synchronization that will take care of the fact that sending a message may take a long time.

尤其是

Optional: implement the sendAll method so that the sending is performed asynchronously w.r.t. the caller (i.e., in a separate thread, which should be started at MsgBatcher creation time and reused for each sending).

根据这个问题的标题

我知道我应该问教授，我已经给他发了邮件，但他没有回复。感谢您的帮助!

Answer 1

我认为您正在准备 PoliMi 的 dist sys 中间件技术考试。

我也在做，我发现了你的问题，因为我正在做同样的练习。这是我的解决方案，看起来效果很好。

我发布了它，即使您可能不再需要它，以防其他人在准备考试时需要它。

import java.util.ArrayList;

public class TestThread extends Thread{

private String name;
private MsgBatcher msgBatcher;

public TestThread(String name, MsgBatcher msgBatcher){
    this.name = name;
    this.msgBatcher = msgBatcher;
}

/**
 * Each thread tries to add 10 messages.
 */
public void run(){
    for(int i = 0; i < 10; i++){
        Message msg = new Message("Thread " + this.name + " | i: " + i);
        try{
            System.out.println("[T] Thread " + this.name + " Adding " + i);
            this.msgBatcher.enqueue(msg);
            System.out.println("[T] Thread " + this.name + " Added " + i);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

public static void main(String[] args){
    MsgBatcher msgBatcher = new MsgBatcher(10);
    TestThread t1 = new TestThread("1", msgBatcher);
    TestThread t2 = new TestThread("2", msgBatcher);

    t1.start();
    t2.start();

    try{
        //After 3 seconds we try to send the first batch
        Thread.sleep(3000);
        msgBatcher.sendAll();
    }catch (Exception e){
        e.printStackTrace();
    }

}
}

class MsgSender extends Thread{

private ArrayList<Message> toSend;

public MsgSender(){
    this.toSend = new ArrayList<Message>();
}

public void run(){
    while(true){
        synchronized (this){
            try{
                System.out.println("[MS] Starting run.");
                while (this.toSend.size() == 0){
                    System.out.println("[MS] Waiting for messages." + this.toSend.size());
                    wait();
                }

                System.out.println("[MS] Messages ready to be sent.");

                for(Message msg: this.toSend){
                    msg.send();
                }

                this.toSend.clear();

                System.out.println("[MS] Messages sent.");

                notifyAll();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

public synchronized void addMsgs(ArrayList<Message> toSend){
    synchronized (this){
        try{
            System.out.println("[MS] Adding messages.");
            while (this.toSend.size() != 0){
                System.out.println("[MS] Adding messages. Waiting toSend to become empty");
                wait();
            }

            this.toSend = toSend;

            System.out.println("[MS] Messages added. Notifying the run."  + this.toSend.size());

            notifyAll();
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    this.toSend = toSend;

    notify();
}
}

class MsgBatcher{

private int size;
private ArrayList<Message> queue;
private MsgSender msgSender;

public MsgBatcher(int size){
    this.size = size;
    this.queue = new ArrayList<Message>(this.size);
    this.msgSender = new MsgSender();
    this.msgSender.start();
    try{
        //wait enough time to be sure the msgSender thread is ready
        Thread.sleep(3000);
    }catch (Exception e){
        e.printStackTrace();
    }
}

public synchronized void enqueue(Message msg) throws InterruptedException{
    System.out.println("[MB] size: " + this.queue.size());
    while(this.queue.size() == this.size){
        wait();
    }

    this.queue.add(msg);

    notifyAll();
}

public synchronized void sendAll(){
    System.out.println("Sending all messages");

    ArrayList<Message> toSend = new ArrayList<Message>();
    toSend.addAll(this.queue);

    this.msgSender.addMsgs(toSend);

    this.queue.clear();

    notifyAll();
}
}

class Message {

private String text;

public Message(String text){
    this.text = text;
}

public void send(){
    System.out.println("[M] Sending: " + this.text);
    try{
        Thread.sleep(1000);
    }catch (Exception e){
        e.printStackTrace();
    }
    System.out.println("[M] Sent: " + this.text);
}
}