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;
}