多线程Java Publisher服务器聊天程序
我正在尝试创建一个聊天应用程序,它有一个发布者、一个服务器和多个订阅者。发布服务器(发送到端口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.*;
公共类出版商{ 公共静态void main(字符串[]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");
}
}
}
}
感谢您的帮助。我只是想弄清楚订户为什么没有收到消息
# 1 楼答案
处理领域时间通信问题有很多可能性。我自己更喜欢使用Events/EventListeners
目前在您的程序中,服务器本身与处理订户连接的线程之间没有通信
还有一个侧节点:即使发布者连接线程和订阅者连接线程之间进行了正确的通信,它现在也无法工作,因为您使用的是相同的
Server
类。这不仅违反了Single-Responsibility-Principle规则,而且还会阻止服务器向订阅者发送消息假设您已经建立了连接,并且您的服务器类现在已与订阅者连接。会发生什么
订户将循环,直到其套接字的输入流上出现消息。很好,这正是我们想要的。但是服务器做什么呢?事实是完全一样的。服务器的
run
方法的try
块中的前几条语句是创建一个BufferedReader并从中读取,直到收到消息为止。现在,我们在每个站点上都有一个套接字,它将无限期地等待某种消息的到来(这显然永远不会发生,因为两个都在等待某样东西)为了防止出现这种情况,您应该先检查流中是否有任何内容需要读取:
现在是第二部分。如果想要在线程之间通信,需要实现一些逻辑。如上所述,我喜欢听众的概念,因此我将展示一个示例,其中我使用了他们:
MessageReceivedListener。java
注意:接口不必扩展
EventListener
,因为EventListener
只是一个标签界面。我自己仍然更喜欢用它来提醒界面的用途服务器。java(节选)
新的构造函数提供了一种将MessageReceivedListener传递给服务器对象的方法。或者你也可以为它创建一个setter
这就是魔法发生的地方。收到消息后,我们只需将其传递给侦听器的
onMessageReceived(String message)
方法。但它到底有什么作用呢?这是我们在创建服务器对象时定义的。 这里有两个例子,一个是匿名类(Java7和更早版本),另一个是lambdas(Java8和更高版本)示例Java 7及更早版本
在这里,我们传递一个anonymous class作为MessageReceivedListener对象,并定义它的行为(在本例中,只调用另一个将处理其余部分的方法)
现在,由于我们的MessageReceivedListener接口只包含一种方法,我们也可以将其视为功能接口,因此使用lambdas来缩短代码并提高可读性
Lambda(Java 8及更高版本)示例
在这个特定的例子中,我们只有一个参数要传递给一个方法,因此可以使用method reference
如何真正实现方法
sendMessageToSubs(String message)
取决于您。但是,您需要跟踪创建了多少具有订户连接的线程,以及希望如何引用它们