java - 如何在生产者和消费者之间共享单个套接字？

Question

如何确保生产者和消费者各自使用相同的套接字与服务器通信？或者，也许另一个类应该处理套接字连接？生产者和消费者都都在客户端上。

似乎套接字连接与Producer和Consumer绑定(bind)在一起，以至于它们看起来无法解开，所以，虽然我想将连接放入第三类，我不知道该怎么做。

我非常喜欢这种生产者和消费者的模式，只是我不知道如何在客户端的约束下实现它 代码，以便与服务器的所有套接字通信都通过单个套接字进行。

我想可能可以发送/检索唯一标识符，但这会使服务器过于复杂。

代码:

package net.bounceme.dur.client;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.bounceme.dur.data.State;
import net.bounceme.dur.data.Title;

public class Producer implements Runnable {

    private final BlockingQueue<Title> queue;
    private final String server = "localhost";
    private final int portNumber = 8080;

    public Producer(BlockingQueue<Title> q) {
        this.queue = q;
    }

    private Title connect() throws IOException, ClassNotFoundException {
        Socket socket = new Socket(server, portNumber);
        Title title = null;
        State state = State.undefined;
        ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
        title = (Title) objectInputStream.readObject();
        return title;
    }

    @Override
    public void run() {
        try {
            connect();
        } catch (IOException | ClassNotFoundException ex) {
            Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
            //exit?
        }
    }
}

消费者:
包net.bounceme.dur.client；

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.bounceme.dur.data.State;
import net.bounceme.dur.data.Title;

public class Consumer implements Runnable {

    private final BlockingQueue<Title> queue;
    private final String server = "localhost";
    private final int portNumber = 8080;

    public Consumer(BlockingQueue<Title> q) {
        this.queue = q;
    }

    private void consume() throws IOException, ClassNotFoundException, InterruptedException {
        Socket socket = new Socket(server, portNumber);
        Title title = queue.take();
        title.setState(State.x);
        ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
        objectOutputStream.writeObject(title);
    }

    @Override
    public void run() {
        try {
            consume();
        } catch (IOException | ClassNotFoundException | InterruptedException ex) {
            Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
            //exit?
        }

    }
}

改编自:

http://www.journaldev.com/1034/java-blockingqueue-example-implementing-producer-consumer-problem

Answer 1

您需要创建一个Socket并共享InputStream和OutputStream，每一个(取决于哪个正在读取以及哪个一个正在写入服务器)。

你的consume()方法也不应该初始化任何东西，它应该准备好一切，以便它可以读取数据并处理它。您的示例代码显示了该方法实际上创建了一个新的 Socket 和一个新的 ObjectOutputStream，而它们应该只创建一次。

Socket s = new Socket(server, port);
InputStream in = s.getInputStream();
OutputStream out = s.getOutputStream();

consumer.setStream(in);     // Or vice-versa
producer.setStream(out);    // You can also provide it as a constructor argument