Scala 聊天应用程序，阻塞问题

Question

我正在用 Scala 编写一个聊天应用程序，问题出在客户端，客户端在将数据发送到回显服务器之前从 StdIn（阻塞）读取，因此如果连接了多个客户端，那么它们就不会t 从服务器接收数据，直到从 StdIn 读取完成。我在想本地 IO，即从 StdIn 和 reading/writing 读取到套接字应该在单独的线程上，但我想不出一种方法来做到这一点，下面是客户端单例代码：

import java.net._
import scala.io._
import java.io._
import java.security._

object Client {

  var msgAcc = ""

  def main(args: Array[String]): Unit = {
    val conn = new ClientConnection(InetAddress.getByName(args(0)), args(1).toInt)
    val server = conn.connect()
    println("Enter a username")
    val user = new User(StdIn.readLine())
    println("Welcome to the chat " + user.username)
    sys.addShutdownHook(this.shutdown(conn, server))
    while (true) {
    val txMsg = StdIn.readLine()//should handle with another thread?
    if (txMsg != null) {
      conn.sendMsg(server, user, txMsg)
      val rxMsg = conn.getMsg(server)
      val parser = new JsonParser(rxMsg)
      val formattedMsg = parser.formatMsg(parser.toJson()) 
      println(formattedMsg)
      msgAcc = msgAcc + formattedMsg + "\n"
      }
    }
  }

  def shutdown(conn: ClientConnection, server: Socket): Unit = {
    conn.close(server)
    val fileWriter = new BufferedWriter(new FileWriter(new File("history.txt"), true))
    fileWriter.write(msgAcc) 
    fileWriter.close()
    println("Leaving chat, thanks for using")
  }

}

下面是 ClientConnection class:

import javax.net.ssl.SSLSocket
import javax.net.ssl.SSLSocketFactory
import javax.net.SocketFactory
import java.net.Socket
import java.net.InetAddress
import java.net.InetSocketAddress
import java.security._
import java.io._
import scala.io._
import java.util.GregorianCalendar
import java.util.Calendar
import java.util.Date
import com.sun.net.ssl.internal.ssl.Provider
import scala.util.parsing.json._

class ClientConnection(host: InetAddress, port: Int) {

  def connect(): Socket = {
    Security.addProvider(new Provider())
    val sslFactory = SSLSocketFactory.getDefault()
    val sslSocket = sslFactory.createSocket(host, port).asInstanceOf[SSLSocket]
    sslSocket
   }

  def getMsg(server: Socket): String = new BufferedSource(server.getInputStream()).getLines().next()

  def sendMsg(server: Socket, user: User, msg: String): Unit = {
    val out = new PrintStream(server.getOutputStream())
    out.println(this.toMinifiedJson(user.username, msg))
    out.flush()
  }  

  private def toMinifiedJson(user: String, msg: String): String = {
    s"""{"time":"${this.getTime()}","username":"$user","msg":"$msg"}"""
  }

  private def getTime(): String = {
    val cal = Calendar.getInstance()
    cal.setTime(new Date())
    "(" + cal.get(Calendar.HOUR_OF_DAY) + ":" + cal.get(Calendar.MINUTE) + ":" + cal.get(Calendar.SECOND) + ")"
  }

  def close(server: Socket): Unit = server.close()
}

这是使用线程从标准输入读取的客户端单例：

import java.net._
import scala.io._
import java.io._
import java.security._
import java.util.NoSuchElementException

object Client {

  var msgAcc = ""

  def main(args: Array[String]): Unit = {
    val conn = new ClientConnection(InetAddress.getByName(args(0)), args(1).toInt)
    val server = conn.connect()
    println("Enter a username")
    val user = new User(StdIn.readLine())
    println("Welcome to the chat " + user.username)
    sys.addShutdownHook(this.shutdown(conn, server))
    new Thread(conn).start()
    while (true) {
    val tx = conn.tx
    if (tx != null) {
      conn.sendMsg(server, user, tx)
      val rxMsg = conn.getMsg(server)
      val parser = new JsonParser(rxMsg)
      val formattedMsg = parser.formatMsg(parser.toJson()) 
      println(formattedMsg)
      msgAcc = msgAcc + formattedMsg + "\n" 
      }
    }
  }

  def shutdown(conn: ClientConnection, server: Socket): Unit = {
    conn.close(server)
    val fileWriter = new BufferedWriter(new FileWriter(new File("history.txt"), true))
    fileWriter.write(msgAcc) 
    fileWriter.close()

这是扩展 Runnable 的 ClientConnection class：

import javax.net.ssl.SSLSocket
import javax.net.ssl.SSLSocketFactory
import javax.net.SocketFactory
import java.net.Socket
import java.net.InetAddress
import java.net.InetSocketAddress
import java.security._
import java.io._
import scala.io._
import java.util.GregorianCalendar
import java.util.Calendar
import java.util.Date
import com.sun.net.ssl.internal.ssl.Provider
import scala.util.parsing.json._

class ClientConnection(host: InetAddress, port: Int) extends Runnable {

  var tx: String = null

  override def run(): Unit = {
     tx = StdIn.readLine()
  }

  def connect(): Socket = {
    Security.addProvider(new Provider())
    val sslFactory = SSLSocketFactory.getDefault()
    val sslSocket = sslFactory.createSocket(host, port).asInstanceOf[SSLSocket]
    sslSocket
   }

  def getMsg(server: Socket): String = new BufferedSource(server.getInputStream()).getLines().next()

  def sendMsg(server: Socket, user: User, msg: String): Unit = {
    val out = new PrintStream(server.getOutputStream())
    out.println(this.toMinifiedJson(user.username, msg))
    out.flush()
  }  

  private def toMinifiedJson(user: String, msg: String): String = {
    s"""{"time":"${this.getTime()}","username":"$user","msg":"$msg"}"""
  }

  private def getTime(): String = {
    val cal = Calendar.getInstance()
    cal.setTime(new Date())
    "(" + cal.get(Calendar.HOUR_OF_DAY) + ":" + cal.get(Calendar.MINUTE) + ":" + cal.get(Calendar.SECOND) + ")"
  }

  def close(server: Socket): Unit = server.close()
}

Answer 1

因此您已成功将输入的读数移至 Runnable，因此它将运行移至另一个 Thread，但现在当我们查看您的逻辑时主线程，我们看到如果不是null，它总是会发送消息。这有几个问题：

你没有在 run 方法中循环，所以你只会收到一条消息，然后你的 run 方法终止 - 你想把它包装在 while(true) 或 while(<some boolean indicating you're not done>) 所以你不断更新它。
仅在向服务器发送一条消息后，您仍在打印来自服务器的消息。您应该解耦它，以便将消息发送到服务器完全在另一个线程上完成。

类似的方法可能会解决问题：

//This is your new run method in your Runnable
override def run(): Unit = {
    while(true) {
        tx = StdIn.readLine()
        conn.sendMsg(server, user, tx) //Note you'll need to pass those references in somehow
    }
}`

然后，在您的主线程中，只需处理获取消息并将它们打印出来：

new Thread(conn).start()
while (true) {
    //note the lack of sending messages in here
    val rxMsg = conn.getMsg(server)
    val parser = new JsonParser(rxMsg)
    val formattedMsg = parser.formatMsg(parser.toJson()) 
    println(formattedMsg)
    msgAcc = msgAcc + formattedMsg + "\n" 
}

这样，两个行为在不同的线程上。

Scala 聊天应用程序，阻塞问题

Scala chat application, blocking issue

java

sockets

multithreading

scala

blocking