如何在生产者和消费者之间共享一个套接字

Question

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

套接字连接似乎是以一种似乎无法解开的方式绑定到Producer和Consumer中的，所以，虽然我想将连接放入第三类，但我不知道如何做到这一点。

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

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

代码：

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