我很高興地宣布 Reactor 的第二個里程碑版本 Reactor，這是我們邁向 1.0 的一步！Reactor 1.0.0.M2 的 Maven Artifacts 已在常用的里程碑儲存庫中提供。

什麼是 Reactor？

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

此版本包含哪些內容？

第二個里程碑版本包含許多錯誤修復和一些真正令人興奮的新功能。Reactor 現在包含一個 Processor 抽象概念，這是一個基於 LMAX Disruptor RingBuffer 的高度最佳化任務處理器。它使用 Reactor 中的通用抽象概念來配置 RingBuffer，並允許您使用 Reactor 的通用 API 而不是 Disruptor 特定的 API。它還刻意跳過 Selector 匹配和 Reactor 提供的動態 Consumer 分配，以便榨取每一滴可能的吞吐量。在 MacBook Pro 上的軼事基準測試顯示，Processor 每秒可以處理約 100,000,000 個事件。是的，您沒看錯：每秒 1 億個事件！

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

最佳化發佈

Reactor 的強大功能之一是 Selector 匹配主題（類似）的發佈/訂閱。它允許您使用主題、匿名物件、可指派的類型階層、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 形式的 mutator，其工作是更新每個事件的資料。如果批次大小大於底層 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 Artifacts

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"
}