Java Publisher Sever聊天程序
Java Publisher Sever chat program
我正在尝试创建一个聊天应用程序，它有一个发布者、一个服务器和多个订阅者。发布者（发送到端口 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 class 发布者 {
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");
}
}
}
}
感谢任何帮助。我只是想弄清楚为什么订阅者没有收到消息
有很多方法可以处理实时通信问题。我自己更喜欢使用 Events / EventListeners。
目前在您的程序中，服务器本身与处理订阅者连接的线程之间没有通信。
也在一个侧节点上：即使发布者连接线程和订阅者连接线程之间有适当的通信，它现在也不会工作，因为您使用的是相同的 Server class。这不仅违反了 Single-Responsibility-Principle，而且还会阻止服务器向订阅者发送消息。
假设您已经建立连接并且您的服务器 class 现在已与订阅者连接。会发生什么？
订阅者将循环，直到在他的套接字的输入流上有一条消息。很好，这正是我们想要的。但是服务器是做什么的呢？事实是一模一样。服务器 run 方法的 try 块中的前几条语句是创建一个 BufferedReader 并从中读取直到收到消息。现在我们在每个站点上都有一个套接字，它将无限地等待某种消息到达（这显然不会发生，因为 both 都在等待某些东西）。
为防止这种情况，您应该首先检查流中是否有任何内容可读：
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 );
}
注意：接口不必扩展 EventListener，因为 EventListener
只是一个标记界面。我自己还是更喜欢用这个来提醒界面的用途。
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) 方法。但它到底做了什么？这是我们在创建服务器对象时定义的。
这里有两个示例，一个使用匿名 classes（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 接口只包含一个方法，我们也可以将其视为功能接口，因此使用 lambda 来缩短代码并提高可读性。
Lambda 示例（Java 8 及更高版本）
Server servpub = new Server( servSockpub.accept(), ++pubNumber, Server::sendMessageToSubscribers);
在这种特定情况下，我们只有一个要传递给方法的参数，因此可以使用 method reference.
如何实际实现方法 sendMessageToSubs(String message) 取决于您。但是您需要跟踪已创建了多少具有订阅者连接的线程以及您希望如何引用它们。
我正在尝试创建一个聊天应用程序，它有一个发布者、一个服务器和多个订阅者。发布者（发送到端口 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 class 发布者 { 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");
}
}
}
}
感谢任何帮助。我只是想弄清楚为什么订阅者没有收到消息
有很多方法可以处理实时通信问题。我自己更喜欢使用 Events / EventListeners。
目前在您的程序中，服务器本身与处理订阅者连接的线程之间没有通信。
也在一个侧节点上：即使发布者连接线程和订阅者连接线程之间有适当的通信，它现在也不会工作，因为您使用的是相同的 Server class。这不仅违反了 Single-Responsibility-Principle，而且还会阻止服务器向订阅者发送消息。
假设您已经建立连接并且您的服务器 class 现在已与订阅者连接。会发生什么？
订阅者将循环，直到在他的套接字的输入流上有一条消息。很好，这正是我们想要的。但是服务器是做什么的呢？事实是一模一样。服务器 run 方法的 try 块中的前几条语句是创建一个 BufferedReader 并从中读取直到收到消息。现在我们在每个站点上都有一个套接字，它将无限地等待某种消息到达（这显然不会发生，因为 both 都在等待某些东西）。
为防止这种情况，您应该首先检查流中是否有任何内容可读：
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 );
}
注意：接口不必扩展 EventListener，因为 EventListener
只是一个标记界面。我自己还是更喜欢用这个来提醒界面的用途。
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) 方法。但它到底做了什么？这是我们在创建服务器对象时定义的。
这里有两个示例，一个使用匿名 classes（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 接口只包含一个方法，我们也可以将其视为功能接口，因此使用 lambda 来缩短代码并提高可读性。
Lambda 示例（Java 8 及更高版本）
Server servpub = new Server( servSockpub.accept(), ++pubNumber, Server::sendMessageToSubscribers);
在这种特定情况下，我们只有一个要传递给方法的参数，因此可以使用 method reference.
如何实际实现方法 sendMessageToSubs(String message) 取决于您。但是您需要跟踪已创建了多少具有订阅者连接的线程以及您希望如何引用它们。