使用Kafka和ZeroMQ实现分布式Quasar Actor
Quasar 是一个类似Scala的Akka的Java Actor模型实现框架，它使用 fiber（纤程）绿色线程实现类似Erlang的Actor模型，本站介绍结合Apache Kafka和ZeroMQ实现Quasar的分布式Actor模型。
当然，Galaxy也是一种非常好的选择，它是一个快速in-memory数据网格，专门为数据本地化复制，可选的持久和分布式actor注册，甚至能够在节点服务器之间迁移actor。
Kafka
Apache Kafka是目前流行的提供事件日志流的项目，它的API包括两种生产者：同步和异步，消费者只有一种：同步。一个Kafka生产者处理是线程安全的，易于使用。
Comsat项目包括一个纤程友好fiber-friendly的 Kafka producer 集成，.我们使用它在actor内实现kafka的生产者，我们案例是演示从生产者发送几千个序列化的消息到消费者。
生产者代码如下：
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(); // 可选，堵塞绿色线程直到记录被持久化，也可以 `producer.flush()`
}
我们使用Comsat的FiberKafkaProducer 包装了KafkaProducer 对象，这是为了获得fiber-blocking的future。
Kafka的消费处理不是线程安全的，且只有线程堵塞的同步方式：
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); // ...
}
}
因为我们并不想堵塞fiber的底层线程池，这里就不能使用FiberAsync.runBlocking来将数据喂给固定大小的线程池了，而是在我们的actor的doRun中，我们使用一个异步任务，它会堵塞fiber会等到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);
}
}
完整案例代码见：complete example，这是一个从生产者发送几千个序列化的消息到消费者。
ØMQ
ØMQ (或 ZeroMQ)是一个非集中的broker解决方案，有各种socket适合不同通讯模式（请求/应答或pub/sub发布/订阅），在我们的案例中我们使用最简单的请求应答模式，也是演示从生产者发送消息到消费者，生产者代码如下：
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
}
这里context实际作为一个socket工厂，其中有I/O线程数量作为context参数，这是因为ZeroMQ的socket不是基于connection-bound OS处理，而是一个简单前端，能够进行连接重试处理，或多个连接处理，或有效率的并发I/O和提供队列。者就是为什么send方法调用几乎从来不会堵塞，而recv方法不是基于连接的I/O调用，而是一个基于你的线程和指定I/O任务之间的同步调用，所谓指定I/O任务是指从一个或多个连接中接受进来的字节数据。
在Actor我们就不使用堵塞fiber的方式了，因为这里send不会堵塞，因此使用FiberAsync.runBlocking堵塞住read调用。
因此改写上面生产者代码如下，这是Actor的代码：
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);
}
}
完整代码见：这里example
更深入应用见：Distributed Quasar Actors with Kafka and ZeroMQ
源码项目： on GitHub