我很高興地宣佈 Reactor 的第二個里程碑版本，它正在朝著 1.0 版本邁進！Reactor 1.0.0.M2 的 Maven 工件可在常規的里程碑儲存庫中獲取。

什麼是 Reactor？

Reactor 是一個基礎框架，用於在 JVM 上構建高吞吐量、非同步、響應式應用程式。它提供 Selector 風格的事件路由主題匹配、動態 Consumer 分配、一個超快的任務處理器以及響應式 Stream 和 Promise API，用於非同步處理資料和協調非同步任務。它提供全面的 Groovy 語言支援，透過提供語言擴充套件，使得用 Groovy 編寫 Reactor 應用程式非常 Groovy！它還具有易於使用的 Spring 支援，可以自動將帶註解的 POJO 連線到 Reactors。

此版本包含哪些內容？

這個第二個里程碑版本包括了一些錯誤修復和一些真正令人興奮的新功能。Reactor 現在包含了一個 Processor 抽象，這是一個基於 LMAX Disruptor RingBuffer 的高度最佳化任務處理器。它使用 Reactor 的通用抽象來配置 RingBuffer，並允許您使用 Reactor 的通用 API，而不是 Disruptor 特定的 API。它還透過設計跳過了 Reactor 提供的 Selector 匹配和動態 Consumer 分配，以榨取其所能達到的每一個吞吐量。在 MacBook Pro 上的傳聞基準測試顯示，Processor 每秒可以透過管道處理大約 1 億個事件。是的，您沒有看錯：每秒 1 億個！

1.0.0.M2 還在 Reactor API 中包含了一個雖小但重要的新功能，它優化了 Reactor 中的事件釋出，以獲得大約 30-50% 的更高吞吐量。它不適用於所有情況，因為它從 Reactor 中準備了一個最佳化的 Consumer 列表，但對於每秒額外的 1000 萬個事件的吞吐量來說，這是一個很棒的新功能。

最佳化釋出

Reactor 的強大之處之一是 Selector 匹配主題（ish）的釋出/訂閱。它允許您使用主題、匿名物件、可分配型別層次結構、URI 路徑匹配或正則表示式（如果您實現自己的領域特定 Selector，則可以使用任何其他型別的 Selector 匹配）輕鬆地將處理程式分配給事件。但是許多應用程式可以在啟動時分配其處理程式，這意味著到這些 Consumer 的路徑可以針對高效的事件釋出進行最佳化。新的 Reactor 方法 prepare(Object) 允許您預先選擇一個鍵的 Consumer。它返回一個 Consumer 本身，事件釋出者可以使用它來有效地通知新事件。

// Create Environment in which Reactors operate
Environment env = new Environment();
Reactor reactor = Reactors.reactor().env(env).get();

reactor.on($("say.hello"), new Consumer<Event<String>>() {
	public void accept(Event<String> ev) {
		System.out.println("Hello " + ev.getData() + "!");
	}
});

Consumer<Event<String>> sayHello = reactor.prepare("say.hello");
for(String name : listOfNames) {
	sayHello.accept(name);
}

RingBuffer 任務處理器

Reactor 1.0.0.M2 包含了 Processor 抽象。它是一個由 LMAX Disruptor RingBuffer 支援的簡單任務處理器，旨在將其無縫整合到 Reactor 中使用的響應式 API 中，因此它使用 Supplier 和 Consumer 等通用抽象。一個完全配置的 Processor 可以在單個表示式中建立，使用 Java 8 lambda 更簡潔。

Processor<Message> proc = new ProcessorSpec<Message>()
	.dataSupplier({ return new Message(); })
	.consume({ msg -> // handle the updated Message object })
	.get();

Processor 提供了兩種與底層 RingBuffer 互動的方式。單操作模式透過呼叫 prepare() 方法從 Processor 請求一個 Operation 物件來工作。Operation 上有一個 get() 方法，用於訪問 RingBuffer 在建立時填充的預分配事件物件。此物件的成員可以用新資料進行更新。準備釋出操作並觸發事件處理程式時，只需呼叫 Operation 的 commit() 方法。

public class Message {
	int type;
	Buffer buffer;
}

@Autowired
Processor<Message> proc;

public void handle(Buffer buff) {
	Operation<Message> op = proc.prepare();

	op.get().type = buff.readInt();
	op.get().buffer = buff;

	op.commit();
}

如果您可以批次操作資料，那麼 Processor 提供了一個 batch(int, Consumer) 方法，允許您指定批處理大小並傳遞一個 Consumer 形式的修改器，其工作是更新每個事件的資料。如果批處理大小大於底層 RingBuffer 的大小，批處理將隱式重新整理，否則釋出步驟將延遲，直到達到批處理大小。這通常會提高吞吐量和效率。

public class Message {
	int type;
	Buffer buffer;
}

@Autowired
Processor<Message> proc;

public void handle(List<Buffer> buffs) {

	proc.batch(buffs.size(), new Consumer<Message>() {
		ListIterator<Buffer> it = buffs.listIterator();

		public void accept(Message msg) {
			Buffer next = it.next();

			msg.type = next.readInt();
			msg.buffer = next;
		}
	});

}

SpringOne2GX

Reactor 將在今年距離不到兩週的 SpringOne2GX 大會上突出展示。將有 Stephane Maldini 和 Jon Brisbin 主導的完整專題討論，以及關於這項技術如何改變您構建應用程式的方式的幾乎不間斷的茶水間討論。仍然有時間 註冊 並預訂房間。但要快！

資源

GitHub：（原始碼，問題跟蹤器）https://github.com/reactor/reactor/
Wiki：https://github.com/reactor/reactor/wiki
API 文件：http://reactor.github.io/docs/api/

Maven 工件

ext {
	reactorVersion = '1.0.0.M2'
}

repositories {
	mavenCentral()
	maven { url 'http://repo.springsource.org/libs-release' }
	maven { url 'http://repo.springsource.org/libs-milestone' }
}

dependencies {
	// Reactor core
	compile "org.projectreactor:reactor-core:$reactorVersion"

	// Reactor Groovy support
	compile "org.projectreactor:reactor-groovy:$reactorVersion"

	// Reactor TCP client/server
	compile "org.projectreactor:reactor-tcp:$reactorVersion"

	// Reactor Spring support
	compile "org.projectreactor:reactor-spring:$reactorVersion"
}