Concurrency Deep Dive:มัลติเธรด

ใน Ruby Magic รุ่นก่อนหน้า เราได้แสดงวิธีใช้งานระบบแชทโดยใช้หลายกระบวนการ ครั้งนี้เราจะแสดงให้คุณเห็นว่าคุณสามารถทำสิ่งเดียวกันโดยใช้หลายเธรดได้อย่างไร

สรุปสั้นๆ

หากคุณต้องการทราบคำอธิบายโดยละเอียดเกี่ยวกับการตั้งค่าพื้นฐาน โปรดอ่านบทความก่อนหน้า แต่เพื่อเตือนคุณอย่างรวดเร็ว:นี่คือลักษณะของระบบแชทของเรา:

เรากำลังใช้ไคลเอนต์เดียวกันกับที่เราใช้ก่อนหน้านี้:

# client.rb
# $ ruby client.rb
require 'socket'
client = TCPSocket.open(ARGV[0], 2000)
 
Thread.new do
  while line = client.gets
    puts line.chop
  end
end
 
while input = STDIN.gets.chomp
  client.puts input
end

การตั้งค่าพื้นฐานสำหรับเซิร์ฟเวอร์จะเหมือนกัน:

# server_threads.rb
# $ ruby server_threads.rb
require 'socket'
 
puts 'Starting server on port 2000'
 
server = TCPServer.open(2000)

ซอร์สโค้ดแบบเต็มที่ใช้ในตัวอย่างในบทความนี้มีอยู่ใน GitHub ดังนั้นคุณจึงทดลองใช้งานได้ด้วยตนเอง

เซิร์ฟเวอร์แชทแบบมัลติเธรด

ตอนนี้เรากำลังเข้าสู่ส่วนที่แตกต่างเมื่อเทียบกับการใช้งานแบบหลายขั้นตอน การใช้ มัลติเธรด เราสามารถทำได้หลายอย่างพร้อมกันด้วยกระบวนการ Ruby เพียงขั้นตอนเดียว เราจะทำสิ่งนี้โดยวางไข่หลายเธรดที่ใช้งานได้

กระทู้

เธรดทำงานอย่างอิสระ โดยรันโค้ดภายในกระบวนการ หลายเธรดสามารถอยู่ในกระบวนการเดียวกันและสามารถแชร์หน่วยความจำได้

<img src="/images/blog/2017-04/threads.png">

จำเป็นต้องใช้ที่เก็บข้อมูลบางส่วนเพื่อจัดเก็บข้อความแชทที่เข้ามา เราจะใช้ Array . ธรรมดา แต่เราต้องการ Mutex . ด้วย เพื่อให้แน่ใจว่ามีเพียงเธรดเดียวเท่านั้นที่เปลี่ยนข้อความพร้อมกัน (เราจะมาดูกันว่า Mutex เป็นอย่างไร ใช้งานได้นิดหน่อย)

mutex = Mutex.new
messages = []

ต่อไปเราจะเริ่มวนรอบที่เราจะยอมรับการเชื่อมต่อที่เข้ามาจากไคลเอนต์แชท เมื่อสร้างการเชื่อมต่อแล้ว เราจะวางเธรดเพื่อจัดการกับข้อความขาเข้าและขาออกจากการเชื่อมต่อไคลเอ็นต์นั้น

Thread.new บล็อคการโทรจนถึง server.accept ส่งคืนบางสิ่งบางอย่าง และป้องกันบล็อกต่อไปนี้ในเธรดที่สร้างขึ้นใหม่ รหัสในชุดข้อความจะอ่านบรรทัดแรกที่ส่งไปและเก็บไว้เป็นชื่อเล่น ในที่สุดก็เริ่มส่งและอ่านข้อความ

loop do
  Thread.new(server.accept) do |socket|
    nickname = read_line_from(socket)
 
    # Send incoming message (coming up)
 
    # Read incoming messages (coming up)
  end
end

มิวเท็กซ์

mutex เป็นอ็อบเจ็กต์ที่อนุญาตให้เธรดหลายเธรดประสานวิธีการใช้ทรัพยากรที่ใช้ร่วมกัน เช่น อาร์เรย์ เธรดสามารถระบุได้ว่าต้องการการเข้าถึง และในช่วงเวลานี้เธรดอื่นไม่สามารถเข้าถึงทรัพยากรที่ใช้ร่วมกันได้

เซิร์ฟเวอร์อ่านข้อความขาเข้าจากซ็อกเก็ต มันใช้ synchronize เพื่อล็อกที่เก็บข้อความ จึงสามารถเพิ่มข้อความไปยังข้อความ Array . ได้อย่างปลอดภัย .

# Read incoming messages
while incoming = read_line_from(socket)
  mutex.synchronize do
    messages.push(
      :time => Time.now,
      :nickname => nickname,
      :text => incoming
    )
  end
end

สุดท้าย Thread ถูกสร้างขึ้นที่ทำงานอย่างต่อเนื่องในลูปเพื่อให้แน่ใจว่าข้อความใหม่ทั้งหมดที่ได้รับจากเซิร์ฟเวอร์ถูกส่งไปยังไคลเอนต์ มันได้รับการล็อคอีกครั้งเพื่อให้รู้ว่าเธรดอื่นไม่รบกวน หลังจากติ๊กลูปเสร็จแล้ว มันก็จะพักสักครู่แล้วไปต่อ

# Send incoming message
Thread.new do
  sent_until = Time.now
  loop do
    messages_to_send = mutex.synchronize do
      get_messages_to_send(nickname, messages, sent_until).tap do
        sent_until = Time.now
      end
    end
    messages_to_send.each do |message|
      socket.puts "#{message[:nickname]}: #{message[:text]}"
    end
    sleep 0.2
  end
end

ล็อกล่ามทั่วโลก

คุณอาจเคยได้ยินเรื่องที่ Ruby ไม่สามารถสร้างเธรด "ของจริง" ได้เนื่องจาก Ruby's Global Interpreter Lock (GIL) นี่เป็นความจริงบางส่วน GIL เป็นตัวล็อครอบการทำงานของรหัส Ruby ทั้งหมด และป้องกันไม่ให้กระบวนการ Ruby ใช้ CPU หลายตัวพร้อมกัน การทำงานของ IO (เช่น การเชื่อมต่อเครือข่ายที่เราใช้ในบทความนี้) ทำงานนอก GIL ซึ่งหมายความว่าคุณสามารถบรรลุการทำงานพร้อมกันที่เหมาะสมได้ในกรณีนี้

สรุป

ตอนนี้ เรามีเซิร์ฟเวอร์แชทที่ทำงานภายในกระบวนการเดียวโดยใช้ athread ต่อการเชื่อมต่อ สิ่งนี้จะใช้ทรัพยากรน้อยกว่าการใช้งานหลายกระบวนการมาก หากคุณต้องการดูรายละเอียดของโค้ดหรือลองใช้งาน คุณสามารถค้นหาโค้ดตัวอย่างได้ที่นี่

ในบทความสุดท้ายในชุดนี้ เราจะใช้เซิร์ฟเวอร์การแชทเดียวกันนี้โดยใช้เธรดเดียวและวนซ้ำของเหตุการณ์ ในทางทฤษฎี วิธีนี้น่าจะใช้ทรัพยากรน้อยกว่าการนำเธรดไปใช้งานด้วยซ้ำ!