通过Java 8流使用Oracle AQ

Oracle数据库最令人敬畏的功能之一是Oracle AQ：Oracle数据库高级队列。 AQ API直接在数据库中实现了完整的事务性消息传递系统。

在数据库位于系统中心的经典体系结构中，使用AQ进行进程间通信时，多个应用程序（其中一些应用程序用Java编写，其他应用程序用Perl或PL / SQL编写）访问同一数据库。太好了如果您更喜欢Java EE，则可以购买基于Java的MQ解决方案，并将该消息总线/中间件放在系统体系结构的中心。但是，为什么不使用数据库呢？

如何在jOOQ中使用PL / SQL AQ API

用于AQ消息入队和出队的PL / SQL API非常简单，可以使用jOOQ的OracleDSL.DBMS_AQ API从Java轻松访问它。

此处使用的队列配置如下所示：

CREATE OR REPLACE TYPE message_t AS OBJECT (ID         NUMBER(7),title      VARCHAR2(100 CHAR)
)
/BEGINDBMS_AQADM.CREATE_QUEUE_TABLE(queue_table => 'message_aq_t',queue_payload_type => 'message_t');DBMS_AQADM.CREATE_QUEUE(queue_name => 'message_q',queue_table => 'message_aq_t');DBMS_AQADM.START_QUEUE(queue_name => 'message_q');COMMIT;
END;
/

并且jOOQ代码生成器将生成有用的类，并将所有类型信息直接与它们相关联（简化示例）：

class Queues {static final Queue<MessageTRecord> MESSAGE_Q = new QueueImpl<>("NEW_AUTHOR_AQ", MESSAGE_T);
}class MessageTRecord {void setId(Integer id) { ... }Integer getId() { ... }void setTitle(String title) { ... }String getTitle() { ... }MessageTRecord(Integer id, String title) { ... }
}

然后，可以使用这些类直接在生成的队列引用上安全地使消息类型入队和出队：

// The jOOQ configuration
Configuration c = ...// Enqueue a message
DBMS_AQ.enqueue(c, MESSAGE_Q, new MessageTRecord(1, "test"));// Dequeue it again
MessageTRecord message = DBMS_AQ.dequeue(c, MESSAGE_Q);

很简单，不是吗？

现在，让我们利用Java 8功能

消息队列就是无限（阻塞）消息流。从Java 8开始，我们为此类消息流提供了强大的API，即Stream API。

这就是为什么我们为即将到来的jOOQ 3.8添加了新的API，将现有的jOOQ AQ API与Java 8 Streams相结合的原因：

// The jOOQ configuration
Configuration c = ...DBMS_AQ.dequeueStream(c, MESSAGE_Q).filter(m -> "test".equals(m.getTitle())).forEach(System.out::println);

上面的流管道将侦听MESSAGE_Q队列，使用所有消息，过滤掉不包含"test"的消息，并打印其余消息。

阻止流

有趣的是，这是一个无限的阻塞流。只要队列中没有新消息，流管道处理就会简单地在队列上阻塞，等待新消息。这对于顺序流来说不是问题，但是在调用Stream.parallel()时会发生什么呢？

jOOQ将消耗事务中的每个消息。 jOOQ 3.8事务在ForkJoinPool.ManagedBlocker运行：

static <T> Supplier<T> blocking(Supplier<T> supplier) {return new Supplier<T>() {volatile T result;@Overridepublic T get() {try {ForkJoinPool.managedBlock(new ManagedBlocker() {@Overridepublic boolean block() {result = supplier.get();return true;}@Overridepublic boolean isReleasable() {return result != null;}});}catch (InterruptedException e) {throw new RuntimeException(e);}return asyncResult;}};
}

这不是很多魔术。当ManagedBlocker由ForkJoinWorkerThread运行时，它会运行一些特殊的代码，以确保线程的ForkJoinPool不会由于线程耗尽而死锁。有关更多信息，请在此处阅读此有趣的文章：http: //zeroturnaround.com/rebellabs/java-parallel-streams-are-bad-for-your-health

或以下堆栈溢出答案： http : //stackoverflow.com/a/35272153/521799

因此，如果您想要超快速的并行AQ出队过程，请运行：

// The jOOQ configuration. Make sure its referenced
// ConnectionPool has enough connections
Configuration c = ...DBMS_AQ.dequeueStream(c, MESSAGE_Q).parallel().filter(m -> "test".equals(m.getTitle())).forEach(System.out::println);

而且您将拥有几个线程，这些线程将使消息并行出队。

不想等待jOOQ 3.8？

没问题。使用当前版本并将dequeue操作包装在您自己的Stream ：

Stream<MessageTRecord> stream = Stream.generate(() ->DSL.using(config).transactionResult(c ->dequeue(c, MESSAGE_Q))
);

做完了

奖励：异步出队

在我们讨论时，排队系统的另一个很好的功能是它们的异步性。在Java 8中， CompletionStage是一个非常有用的用于建模（和组合）异步算法的类型，它是默认实现CompletableFuture ，它再次在ForkJoinPool执行任务。

使用jOOQ 3.8，您可以再次简单地调用

// The jOOQ configuration. Make sure its referenced
// ConnectionPool has enough connections
Configuration c = ...CompletionStage<MessageTRecord> stage =
DBMS_AQ.dequeueAsync(c, MESSAGE_Q).thenCompose(m -> ...)...;

敬请期待jOOQ博客上的另一篇文章，我们将研究更复杂的异步用例，并使用jOOQ 3.8和Java 8阻止SQL语句。