Java Многопоточность производитель-потребитель

Question

Следуя этому коду, как реализовать многопоточность между записями отправки и получения.

public class Test_ProducerConsumer {  

    // Main method
    public static void main(String[] args) {

        // Send records
        for(int i=0; i<10; i++) {
            kafka_io.send_records(sendtopic, "test_key", "test_value" + String.valueOf(i));         
        }

        // Get records
        try {
            kafka_io.get_records(gettopic, 100000);
        } catch (JSONException e) {
            e.printStackTrace();
        } 
    }
}

Answer 1

Вы можете использовать BlockingQueue .

На стороне получения элементов вы можете заблокировать поток, пока не появится новый элемент, в то время как с любым другим потоком вы можете добавить элемент и уведомить поток, который получает элементы.

// Inside a class
public static void main(String[] args) {
    Runnable r = Consumer()
    Thread t = Thread(r)
    t.start()

    // Send records
    for(int i=0; i<10; i++) {
        r.queue.add(DataWrapper(...));         
    }
}

class Consumer implements Runnable {
    BlockingQueue<DataWrapper> queue = BlockingQueue();
    @Override
    public void run() {
        while(true) {
            DataWrapper data = queue.take();
            // use data here
        }
    }
}

class DataWrapper {
    String dataString;   // these are your variables change as you want
    int dataInt;
}