java - Java 中 Go channel 的等价物

Question

我有一个要求，我需要从一组阻塞队列中读取数据。阻塞队列是由我正在使用的库创建的。我的代码必须从队列中读取。我不想为这些阻塞队列中的每一个创建一个读取器线程。相反，我想使用单个线程(或者可能最多使用 2/3 个线程)来轮询它们的数据可用性。由于一些阻塞队列可能长时间没有数据，而其中一些可能会出现数据突发。用小超时轮询队列会起作用，但这根本不是有效的，因为即使其中一些队列长时间没有数据，它仍然需要继续循环遍历所有队列。基本上，我正在寻找一种阻塞队列的选择/epoll(用于套接字)机制。任何线索都非常感谢。

不过，在 Go 中做到这一点真的很容易。下面的代码对 channel 和 goroutine 进行了同样的模拟:

package main

import "fmt"
import "time"
import "math/rand"

func sendMessage(sc chan string) {
    var i int

    for {
        i =  rand.Intn(10)
        for ; i >= 0 ; i-- {
            sc <- fmt.Sprintf("Order number %d",rand.Intn(100))
        }
        i = 1000 + rand.Intn(32000);
        time.Sleep(time.Duration(i) * time.Millisecond)
    }
}

func sendNum(c chan int) {
    var i int 
    for  {
        i = rand.Intn(16);
        for ; i >=  0; i-- {
            time.Sleep(20 * time.Millisecond)
            c <- rand.Intn(65534)
        }
        i = 1000 + rand.Intn(24000);
        time.Sleep(time.Duration(i) * time.Millisecond)
    }
}

func main() {
    msgchan := make(chan string, 32)
    numchan := make(chan int, 32)
    i := 0
    for ; i < 8 ; i++ {
        go sendNum(numchan)
        go sendMessage(msgchan)
    }
    for {
        select {
        case msg := <- msgchan:
            fmt.Printf("Worked on  %s\n", msg)
        case x := <- numchan:
            fmt.Printf("I got %d \n", x)
        }
    }
}

Answer 1

我建议您考虑使用 JCSP图书馆。 Go 的 select 的等效项称为 Alternative .您只需要一个消费线程，如果它使用 Alternative 打开它们，则不需要轮询传入的 channel 。因此，这将是复用源数据的一种有效方式。

如果您能够将 BlockingQueues 替换为 JCSP channel ，将会有很大帮助。 channel 的行为基本相同，但在 channel 端共享的扇出或扇入方面提供了更大程度的灵 active ，特别是在 Alternative 中使用 channel 。

作为一个使用示例，这里是一个公平的多路复用器。此示例演示了一个过程，该过程将流量从其输入 channel 阵列公平地多路复用到其单个输出 channel 。无论竞争对手多么渴望，任何输入 channel 都不会被饿死。

import org.jcsp.lang.*;

public class FairPlex implements CSProcess {

   private final AltingChannelInput[] in;
   private final ChannelOutput out;

   public FairPlex (final AltingChannelInput[] in, final ChannelOutput out) {
     this.in = in;
     this.out = out;
   }

   public void run () {

     final Alternative alt = new Alternative (in);

     while (true) {
       final int index = alt.fairSelect ();
       out.write (in[index].read ());
     }
   }
 }

请注意，如果上面使用了 priSelect，那么如果索引较低的 channel 不断要求服务，那么索引较高的 channel 就会被饿死。或者，可以使用 select 代替 fairSelect，但这样就无法进行饥饿分析。 select 机制应该只在饥饿不是问题时使用。

摆脱僵局

与 Go 一样，使用 channel 的 Java 程序必须设计为不会出现死锁。在 Java 中实现低级并发原语非常困难，您需要一些可靠的东西。幸运的是，Alternative 与 JCSP channel 一起通过了形式分析的验证。这使它成为一个可靠的选择。

为了澄清一点困惑，当前的 JCSP 版本是 1.1-rc5在 Maven 存储库中，而不是网站上所说的。