🎓 A1Learner
🔍
0 of 45 lessons completed

Channels at a First Look

A channel is a queue that moves values from a producer to a consumer. The producer sends values and the consumer reads them in order.

A channel moves numbers in order

The producer sends three numbers and then closes the channel. The consumer collects them in order.

import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
@main def main(): Unit =
  val channel = LinkedBlockingQueue[Option[Int]]()
  val collected = ListBuffer.empty[Int]
  val producer = new Thread(() => {
    channel.put(Some(1))
    channel.put(Some(2))
    channel.put(Some(3))
    channel.put(None)
  })
  val consumer = new Thread(() => {
    var running = true
    while (running) {
      channel.take() match
        case Some(v) => collected += v
        case None => running = false
    }
  })
  producer.start()
  consumer.start()
  producer.join()
  consumer.join()
  println(collected.toList)

A channel can carry larger values

The producer sends two larger numbers and then closes the channel. The consumer collects both.

import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
@main def main(): Unit =
  val channel = LinkedBlockingQueue[Option[Int]]()
  val collected = ListBuffer.empty[Int]
  val producer = new Thread(() => {
    channel.put(Some(40))
    channel.put(Some(50))
    channel.put(None)
  })
  val consumer = new Thread(() => {
    var running = true
    while (running) {
      channel.take() match
        case Some(v) => collected += v
        case None => running = false
    }
  })
  producer.start()
  consumer.start()
  producer.join()
  consumer.join()
  println(collected.toList)

A consumer can change each value

The producer sends 2 and 4. The consumer doubles each value before collecting it.

import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
@main def main(): Unit =
  val channel = LinkedBlockingQueue[Option[Int]]()
  val collected = ListBuffer.empty[Int]
  val producer = new Thread(() => {
    channel.put(Some(2))
    channel.put(Some(4))
    channel.put(None)
  })
  val consumer = new Thread(() => {
    var running = true
    while (running) {
      channel.take() match
        case Some(v) => collected += v * 2
        case None => running = false
    }
  })
  producer.start()
  consumer.start()
  producer.join()
  consumer.join()
  println(collected.toList)

Example

import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
@main def main(): Unit =
  val channel = LinkedBlockingQueue[Option[Int]]()
  val collected = ListBuffer.empty[Int]
  val producer = new Thread(() => {
    channel.put(Some(10))
    channel.put(Some(20))
    channel.put(Some(30))
    channel.put(None)
  })
  val consumer = new Thread(() => {
    var running = true
    while (running) {
      channel.take() match
        case Some(v) => collected += v
        case None => running = false
    }
  })
  producer.start()
  consumer.start()
  producer.join()
  consumer.join()
  println(collected.toList)
  println(collected.sum)

Key points

  • Producer: A producer is the code that sends values into the channel.
  • Consumer: A consumer is the code that reads values out of the channel.
  • Close value: A close value is a special value that means no more values will arrive.
  • Use a LinkedBlockingQueue with a close value to build a small channel by hand.
  • Order: Values leave the channel in the same order the producer sent them.

In this example, the producer sends 10, 20 and 30 and then closes the channel. The consumer collects the values in order, and the program prints the collected list and then their sum.

Note

This lesson builds a channel by hand, while real channel libraries add features such as buffering choices and selection.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Scastie

Exercise

What tells the consumer that no more values will arrive?