Java Publisher Server 聊天程序

Question

我正在尝试创建一个聊天应用程序，它有一个发布者、一台服务器和多个订阅者。发布者(发送到端口 8000)向服务器(监听端口 8000 和 5000)发送消息，服务器将其进一步转发给订阅者(监听端口 5000)。

现在到目前为止，我可以创建多个发布者，并且服务器和发布者之间的通信正常，但是，我无法将发布者发送的消息发送给订阅者

服务器端代码

package serverclient;
import java.io.*;
import java.net.*;
import java.util.logging.Level;
import java.util.logging.Logger;
public class Server extends Thread{
private Socket socket;
private int clientNumber;

public Server(Socket socket, int clientNumber){
    this.socket  = socket;
    this.clientNumber = clientNumber;
    if(socket.getLocalPort() == 5000)System.out.print("\nSubscriber "+ clientNumber +" is connected to the server");
    if(socket.getLocalPort() == 8000)System.out.print("\nPublisher "+ clientNumber +" is connected to the server");
} 

@Override
public void run(){
    try {
        BufferedReader dStream = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        PrintWriter out = new PrintWriter(socket.getOutputStream(), true);

        while(true){
            synchronized(this){
                String clMessage = dStream.readLine();
                System.out.println("\n"+clMessage);
                // if(socket.getLocalPort() == 5000){
                    out.println("Hey the server is sending the message to subscriber");
                // }    
                //out.println("Hey the publisher has sent the message :  " + clMessage);
            }
        }

    } catch (IOException ex) {
        System.out.print("\nError has been handled 1\n");
    }finally{
        try {
            socket.close();
        } catch (IOException ex) {
            System.out.print("\nError has been handled 2\n");
        }
    }

}

public static void main(String [] args) throws IOException{
    int subNumber = 0;
    int pubNumber = 0;
    ServerSocket servSockpub = new ServerSocket(8000);   
    ServerSocket servSocksub = new ServerSocket(5000);
    try {
        while (true) {
            Server servpub = new Server(servSockpub.accept(),++pubNumber);
            servpub.start();
            System.out.print("\nThe server is running on listen port "+ servSockpub.getLocalPort());
            Server servsub = new Server(servSocksub.accept(),++subNumber);
            servsub.start();
            System.out.print("\nThe server is running on listen port "+ servSocksub.getLocalPort());
        }
    } finally {
        servSockpub.close();
        servSocksub.close();
    }
}

 }

发布商代码

 package serverclient;
 import java.net.*;
 import java.io.*;

公共(public)类发布者{ public static void main(String [] args) 抛出 IOException{

    Socket sock = new Socket("127.0.0.1",8000);

    // reading from keyboard (keyRead object)
    BufferedReader keyRead = new BufferedReader(new InputStreamReader(System.in));

    // sending to client (pwrite object)
    OutputStream ostream = sock.getOutputStream(); 
    PrintWriter pwrite = new PrintWriter(ostream, true);

    InputStream istream = sock.getInputStream();
    BufferedReader receiveRead = new BufferedReader(new InputStreamReader(istream));

    System.out.println("Start the chitchat, type and press Enter key");

    String receiveMessage,sendMessage;               
    while(true)
    {
        sendMessage = keyRead.readLine();  // keyboard reading
        pwrite.println(sendMessage);       // sending to server
        pwrite.flush();                    // flush the data  

        if((receiveMessage = receiveRead.readLine()) != null) //receive from server
        {
            System.out.println(receiveMessage); // displaying at DOS prompt
        }  
        else{
            System.out.print("Null");
        }
    }

}
   }

订阅者

    package serverclient;

    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.InputStreamReader;
    import java.io.OutputStream;
    import java.io.PrintWriter;
    import java.net.Socket;


   public class Subscriber {
    public static void main (String [] args) throws IOException{

    Socket sock = new Socket("127.0.0.1",5000);

  // receiving from server ( receiveRead  object)
    InputStream istream = sock.getInputStream();
    BufferedReader receiveRead = new BufferedReader(new InputStreamReader(istream));

    System.out.println("Recive side");

    String receiveMessage, sendMessage;  
    while(true)
    {
        System.out.print("Hey man " + receiveRead.readLine() + "\n");
        if((receiveMessage = receiveRead.readLine()) != null) //receive from server
        {
            System.out.println(receiveMessage); // displaying at DOS prompt
        }  
        else{
            System.out.print("Null");
        }
    }

}

}

感谢任何帮助。我只是想弄清楚为什么订阅者没有收到消息

Answer 1

处理实时通信问题有很多可能性。我自己更喜欢使用事件/事件监听器。
目前，在您的程序中，服务器本身与处理订阅者连接的线程之间没有通信。

还在侧节点上:即使发布者连接线程和订阅者连接线程之间进行了正确的通信，它现在也无法工作，因为您使用的是相同的 Server 类。这不仅违反了Single-Responsibility-Principle但也会阻止服务器向订阅者发送消息。

假设您已经建立了连接，并且您的服务器类现在已与订阅者连接。会发生什么？
订阅者将循环直到其套接字的输入流上有消息。很好，这正是我们想要的。但是服务器做什么呢？事实是完全一样。服务器的 run 方法的 try block 中的前几条语句是创建一个 BufferedReader 并从中读取数据，直到收到消息。现在我们在每个站点上都有一个套接字，它将无限地等待某种消息的到达(这显然永远不会发生，因为两者都在等待某些东西)。

为了防止这种情况，您应该首先检查流上是否有任何内容可以读取:

while ( true )
{
    if ( socket.getInputStream().available() != 0 )
    {
      // reading logic goes here.... 
      synchronized ( this )
      {
            String clMessage = dStream.readLine();
            System.out.println( "\n" + clMessage );
            out.println( "Hey the server is sending the message to subscriber" );
       }
     }
     // what shall be done when not reading.
}

<小时/>

现在是第二部分。如果您想在线程之间进行通信，您需要实现一些逻辑来执行此操作。如上所述，我喜欢监听器的概念，因此我将展示一个使用它们的示例:

MessageReceivedListener.java

import java.util.EventListener;

public interface MessageReceivedListener
    extends EventListener
{

  public void onMessageReceived( String message );

}

注意:该接口(interface)不必扩展 EventListener，因为 EventListener只是一个标记接口(interface)。我自己仍然更喜欢用它来提醒界面的用途。

Server.java(摘录)

// New constructor since we will pass a Listener now. Also new local variable for it.
public Server( Socket socket, int clientNumber, MessageReceivedListener mrl )
  {
    this.socket = socket;
    this.clientNumber = clientNumber;
    this.mrl = mrl;
    if ( socket.getLocalPort() == 5000 )
      System.out.print( "\nSubscriber " + clientNumber + " is connected to the server" );
    if ( socket.getLocalPort() == 8000 )
      System.out.print( "\nPublisher " + clientNumber + " is connected to the server" );
  }

新的构造函数提供了一种将 MessageReceivedListener 传递给 Server 对象的方法。或者，您也可以为其创建一个 setter 。

synchronized ( this )
      {
        String clMessage = dStream.readLine();
        System.out.println( "\n" + clMessage );
        out.println( "Hey the server is sending the message to subscriber" );
        mrl.onMessageReceived( clMessage );
      }

这就是奇迹发生的地方。收到消息后，我们只需将其传递给监听器的 onMessageReceived(String message) 方法。但它到底有什么作用呢？这是我们在创建服务器对象时定义的。下面是两个示例，一个使用匿名类(Java 7 及之前版本)，另一个使用 lambda(Java 8 及更高版本)。

Java 7 及更早版本示例

Server servpub = new Server( servSockpub.accept(), ++pubNumber,
            new MessageReceivedListener()
            {

              @Override
              public void onMessageReceived( String message )
              {
                // call nother local method
                // this method would need to be a static method of Server
                // because it's in the scope of your server class
                sendMessageToSubscribers(message);
              }
            } );

这里我们传递一个anonymous class作为我们的 MessageReceivedListener 对象并定义它的行为(在本例中只需调用另一个方法来处理其余的事情。

<小时/>

现在，由于我们的 MessageReceivedListener 接口(interface)仅包含一种方法，我们也可以将其视为函数式接口(interface)，因此使用 lambda 来缩短代码并提高可读性。

Lambda 示例(Java 8 及更高版本)

Server servpub = new Server( servSockpub.accept(), ++pubNumber, Server::sendMessageToSubscribers);

在这种特定情况下，我们只有一个参数要传递给方法，因此可以使用 method reference .

如何实际实现方法sendMessageToSubs(String message)取决于您。但是您需要跟踪已创建了多少个具有订阅者连接的线程以及您希望如何引用它们。