带有Kafka和ZeroMQ的分布式类星体演员

因此，您已经有了使用actor的精美设计，选择了JVM和Quasar在该主题上的强大而忠实的观点。所有明智的决定，但是在集群上进行分配时您有什么选择呢？

星系

Galaxy是一个非常酷的选择：快速的内存中数据网格，针对数据局部性进行了优化，具有复制，可选的持久性，分布式参与者注册表，甚至参与者之间的参与者迁移！只有一个警告：要发布正式版的生产质量的银河版，还需要几个月的时间。不建议将当前版本的Galaxy用于生产。

如果我们需要在那之前上线怎么办？

幸运的是，Quasar Actors的阻塞编程模型非常简单，以至于将其与大多数消息传递解决方案集成起来都是轻而易举的，并且为了证明让我们用两种快速，流行且截然不同的模型来做到这一点： Apache Kafka和ØMQ 。

代码和计划

可以在GitHub上找到以下所有示例，只需快速阅读简短的README ，即可立即运行它们。

Kafka和ØMQ分别有两个示例：

快速而肮脏的人直接进行发布/投票或发送/接收演员的呼叫。
详细介绍了代理角色，这些代理角色将您的代码与消息传递API隔离开。为了证明我没有在说谎，该程序对两种技术使用了相同的生产者和消费者参与者类 ，并且几乎使用了相同的引导程序。

卡夫卡

Apache Kafka的采用率急剧上升，这是由于其基于持久性提交日志和用于并行消息使用的使用者组的独特设计：这种结合形成了快速，可靠，灵活和可扩展的代理。

该API包括两种类型的生产者：sync和async；一种消费者（仅sync）； Comsat包括社区贡献的，对光纤友好的Kafka生产商集成。

Kafka生产者句柄是线程安全的，在共享时表现最佳，并且可以像这样在actor主体（或其他任何地方）中轻松获得和使用：

final Properties producerConfig = new Properties();
producerConfig.put("bootstrap.servers", "localhost:9092");
producerConfig.put("client.id", "DemoProducer");
producerConfig.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
producerConfig.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");try (final FiberKafkaProducer<Integer, byte[]> producer = new FiberKafkaProducer<>(new KafkaProducer<>(producerConfig))) {final byte[] myBytes = getMyBytes(); // ...final Future<RecordMetaData> res = producer.send(new ProducerRecord<>("MyTopic", i, myBytes));res.get(); // Optional, blocks the fiber until the record is persisted; thre's also `producer.flush()`
}

我们用Comsat的FiberKafkaProducer包装了KafkaProducer对象，以便找回光纤阻塞的未来。

但是，使用者句柄不是线程安全的^1，并且仅是线程阻塞的：

final Properties producerConfig = new Properties();
consumerConfig = new Properties();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
consumerConfig.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
consumerConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");try (final Consumer<Integer, byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {consumer.subscribe(Collections.singletonList(TOPIC));final ConsumerRecords<Integer, byte[]> records = consumer.poll(1000L);for (final ConsumerRecord<Integer, byte[]> record : records) {final byte[] v = record.value();useMyBytes(v); // ...}
}

由于我们不想阻塞光纤的基础线程池（除了卡夫卡在doRun的线程池，我们无法对其做太多的事情），因此在我们的actor的doRun我们将使用FiberAsync.runBlocking代替FiberAsync.runBlocking来喂入固定的FiberAsync.runBlocking具有异步任务的size执行程序，该任务将阻塞光纤直到poll （将在给定的池中执行）返回之前：

final ExecutorService e = Executors.newFixedThreadPool(2);try (final Consumer<Integer, byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {consumer.subscribe(Collections.singletonList(TOPIC));final ConsumerRecords<Integer, byte[]> records = call(e, () -> consumer.poll(1000L));for (final ConsumerRecord<Integer, byte[]> record : records) {final byte[] v = record.value();useMyBytes(v); // ...}
}

其中call是一个定义如下的实用程序方法（如果不是此Java编译器bug，则没有必要）：

@Suspendable
public static <V> V call(ExecutorService es, Callable<V> c) throws InterruptedException, SuspendExecution {try {return runBlocking(es, (CheckedCallable<V, Exception>) c::call);} catch (final InterruptedException | SuspendExecution e) {throw e;} catch (final Exception e) {throw new RuntimeException(e);}
}

在第一个完整的示例中，我们将从生产者角色向消费者发送一千个序列化消息。

ØMQ

ØMQ（或ZeroMQ）不是集中的代理解决方案，而更多地是各种通信模式（请求/答复，发布/订阅等）的套接字的一般化。在我们的示例中，我们将使用最简单的请求-答复模式。这是我们的新生产者代码：

try (final ZMQ.Context zmq = ZMQ.context(1 /* IO threads */);final ZMQ.Socket trgt = zmq.socket(ZMQ.REQ)) {trgt.connect("tcp://localhost:8000");final byte[] myBytes = getMyBytes(); // ...trgt.send(baos.toByteArray(), 0 /* flags */)trgt.recv(); // Reply, e.g. ACK
}

如您所见，上下文充当套接字工厂，并传递了要使用的I / O线程数：这是因为ØMQ套接字不是连接绑定的OS句柄，而是用于处理的机器的简单前端重试连接，多个连接，高效的并发I / O甚至为您排队。这就是为什么send调用几乎永远不会阻塞，而recv调用不是连接上的I / O调用，而是线程与专门的I / O任务之间的同步的原因，该任务将从一个或多个连接中传入字节。

不过，我们将在角色中阻塞光纤，而不是线程，因此让我们在read调用上使用FiberAsync.runBlocking ，以防万一它阻塞甚至在send时阻塞：

final ExecutorService ep = Executors.newFixedThreadPool(2);try (final ZMQ.Context zmq = ZMQ.context(1 /* IO threads */);final ZMQ.Socket trgt = zmq.socket(ZMQ.REQ)) {exec(e, () -> trgt.connect("tcp://localhost:8000"));final byte[] myBytes = getMyBytes(); // ...call(e, trgt.send(myBytes, 0 /* flags */));call(e, trgt::recv); // Reply, e.g. ACK
}

这是消费者：

try (final ZMQ.Context zmq = ZMQ.context(1 /* IO threads */);final ZMQ.Socket src = zmq.socket(ZMQ.REP)) {exec(e, () -> src.bind("tcp://*:8000"));final byte[] v = call(e, src::recv);exec(e, () -> src.send("ACK"));useMyBytes(v); // ...
}

exec是另一个实用程序函数，类似于call ：

@Suspendable
public static void exec(ExecutorService es, Runnable r) throws InterruptedException, SuspendExecution {try {runBlocking(es, (CheckedCallable<Void, Exception>) () -> { r.run(); return null; });} catch (final InterruptedException | SuspendExecution e) {throw e;} catch (final Exception e) {throw new RuntimeException(e);}
}

这是完整的第一个示例。

在不改变逻辑的情况下进行分配：与救援人员的松散耦合

很简单，不是吗？但是，有些令人讨厌的事情：我们与网络另一端的参与者打交道的方式与本地参与者不同。无论他们位于何处或如何连接，这些都是我们愿意写的演员：

public final class ProducerActor extends BasicActor<Void, Void> {private final ActorRef<Msg> target;public ProducerActor(ActorRef<Msg> target) {this.target = target;}@Overrideprotected final Void doRun() throws InterruptedException, SuspendExecution {for (int i = 0; i < MSGS; i++) {final Msg m = new Msg(i);System.err.println("USER PRODUCER: " + m);target.send(m);}System.err.println("USER PRODUCER: " + EXIT);target.send(EXIT);return null;}
}

public final class ConsumerActor extends BasicActor<Msg, Void> {@Overrideprotected final Void doRun() throws InterruptedException, SuspendExecution {for (;;) {final Msg m = receive();System.err.println("USER CONSUMER: " + m);if (EXIT.equals(m))return null;}}
}

幸运的是，每个演员，无论做什么，都具有相同的非常基本的接口：传入消息队列，称为信箱。这意味着我们可以在两个通信参与者之间插入任意数量的中间参与者或代理，尤其是我们希望有一个发送代理，它将通过中间件将消息获取到目标主机，并在其中接收接收代理，以捕获传入的消息。并将它们放入目标目的地的邮箱中。

因此，在我们的主程序中，我们将为我们的ProducerActor提供合适的发送代理，然后让ConsumerActor从合适的接收代理接收：

final ProducerActor pa = Actor.newActor(ProducerActor.class, getSendingProxy()); // ...
final ConsumerActor ca = Actor.newActor(ConsumerActor.class);
pa.spawn();
System.err.println("USER PRODUCER started");
subscribeToReceivingProxy(ca.spawn()); // ...
System.err.println("USER CONSUMER started");
pa.join();
System.err.println("USER PRODUCER finished");
ca.join();
System.err.println("USER CONSUMER finished");

让我们看看如何首先使用Kafka然后使用ØMQ来实现这些代理。

卡夫卡男演员代理

代理参与者的工厂将与特定的Kafka主题相关联：这是因为可以对主题进行分区，以使多个使用者可以同时读取该主题。我们希望能够最佳地利用每个主题的最大级别或并发性：

/* ... */ KafkaProxies implements AutoCloseable {/* ... */ KafkaProxies(String bootstrap, String topic) { /* ... */ }// ...
}

当然，我们希望对多个参与者使用一个主题，因此发送代理将指定接收者参与者ID，接收代理将仅将消息转发给绑定到该ID的用户参与者：

/* ... */ <M> ActorRef<M> create(String actorID) { /* ... */ }
/* ... */ void drop(ActorRef ref) throws ExecutionException, InterruptedException { /* ... */ }
/* ... */ <M> void subscribe(ActorRef<? super M> consumer, String actorID) { /* ... */ }
/* ... */ void unsubscribe(ActorRef<?> consumer, String actorID) { /* ... */ }

关闭AutoClosable工厂将告诉所有代理终止，并清理簿记参考：

/* ... */ void close() throws Exception { /* ... */ }

生产者实现是非常简单和无趣的，同时给消费者带来了更多的乐趣，因为它将使用Quasar Actors的选择性接收将传入消息保留在其邮箱中，直到至少有一个订阅的用户actor可以使用它们为止：

@Override
protected Void doRun() throws InterruptedException, SuspendExecution {//noinspection InfiniteLoopStatementfor (;;) {// Try extracting from queuefinal Object msg = tryReceive((Object m) -> {if (EXIT.equals(m))return EXIT;if (m != null) {//noinspection uncheckedfinal ProxiedMsg rmsg = (ProxiedMsg) m;final List<ActorRef> l = subscribers.get(rmsg.actorID);if (l != null) {boolean sent = false;for (final ActorRef r : l) {//noinspection uncheckedr.send(rmsg.payload);sent = true;}if (sent) // Someone was listening, remove from queuereturn m;}}return null; // No subscribers (leave in queue) or no messages});// Something from queueif (msg != null) {if (EXIT.equals(msg)) {return null;}continue; // Go to next cycle -> precedence to queue}// Try receiving//noinspection Convert2Lambdafinal ConsumerRecords<Void, byte[]> records = call(e, () -> consumer.get().poll(100L));for (final ConsumerRecord<Void, byte[]> record : records) {final byte[] v = record.value();try (final ByteArrayInputStream bis = new ByteArrayInputStream(v);final ObjectInputStream ois = new ObjectInputStream(bis)) {//noinspection uncheckedfinal ProxiedMsg rmsg = (ProxiedMsg) ois.readObject();final List<ActorRef> l = subscribers.get(rmsg.actorID);if (l != null && l.size() > 0) {for (final ActorRef r : l) {//noinspection uncheckedr.send(rmsg.payload);}} else {ref().send(rmsg); // Enqueue}} catch (final IOException | ClassNotFoundException e) {e.printStackTrace();throw new RuntimeException(e);}}
}

由于我们还需要处理邮箱，因此我们以足够小的超时来轮询Kafka。还要注意，许多参与者可以订阅相同的ID，传入的消息将广播给所有参与者。每个主题创建的接收actor代理（即光纤）的数量，以及池线程和Kafka使用者句柄（ consumer是本地线程，因为Kafka使用者不是线程安全的）的数量将等于每个主题的分区数：这使接收吞吐量达到最大。

目前，此实现使用Java序列化在字节之间来回转换消息，但是当然可以使用其他框架，例如Kryo 。

ØMQ演员代理

ØMQ模型是完全去中心化的：既没有经纪人，也没有话题，因此我们可以简单地将ØMQ地址/端点与一组参与者等同，而无需使用额外的参与者ID：

/* ... */ ZeroMQProxies implements AutoCloseable {/* ... */ ZeroMQProxies(int ioThreads) { /* ... */ }/* ... */ <M> ActorRef<M> to(String trgtZMQAddress) { /* ... */ }/* ... */ void drop(String trgtZMQAddress)/* ... */ void subscribe(ActorRef<? super M> consumer, String srcZMQEndpoint) { /* ... */ }/* ... */ void unsubscribe(ActorRef<?> consumer, String srcZMQEndpoint) { /* ... */ }/* ... */ void close() throws Exception { /* ... */ }
}

同样，在这种情况下，并且由于与以前相同的原因，使用者有点有趣，但幸运的是，线程安全性方面的任何问题都因为ØMQ套接字在多个线程中可以正常工作：

@Override
protected Void doRun() throws InterruptedException, SuspendExecution {try(final ZMQ.Socket src = zmq.socket(ZMQ.REP)) {System.err.printf("PROXY CONSUMER: binding %s\n", srcZMQEndpoint);Util.exec(e, () -> src.bind(srcZMQEndpoint));src.setReceiveTimeOut(100);//noinspection InfiniteLoopStatementfor (;;) {// Try extracting from queuefinal Object m = tryReceive((Object o) -> {if (EXIT.equals(o))return EXIT;if (o != null) {//noinspection uncheckedfinal List<ActorRef> l = subscribers.get(srcZMQEndpoint);if (l != null) {boolean sent = false;for (final ActorRef r : l) {//noinspection uncheckedr.send(o);sent = true;}if (sent) // Someone was listening, remove from queuereturn o;}}return null; // No subscribers (leave in queue) or no messages});// Something processable is thereif (m != null) {if (EXIT.equals(m)) {return null;}continue; // Go to next cycle -> precedence to queue}System.err.println("PROXY CONSUMER: receiving");final byte[] msg = Util.call(e, src::recv);if (msg != null) {System.err.println("PROXY CONSUMER: ACKing");Util.exec(e, () -> src.send(ACK));final Object o;try (final ByteArrayInputStream bis = new ByteArrayInputStream(msg);final ObjectInputStream ois = new ObjectInputStream(bis)) {o = ois.readObject();} catch (final IOException | ClassNotFoundException e) {e.printStackTrace();throw new RuntimeException(e);}System.err.printf("PROXY CONSUMER: distributing '%s' to %d subscribers\n", o, subscribers.size());//noinspection uncheckedfor (final ActorRef s : subscribers.getOrDefault(srcZMQEndpoint, (List<ActorRef>) Collections.EMPTY_LIST))//noinspection uncheckeds.send(o);} else {System.err.println("PROXY CONSUMER: receive timeout");}}}
}