ZeroMQ Support

Spring Integration 提供组件以支持应用程序中的 ZeroMQ 通信。该实现基于 JeroMQ 库的良好支持的 Java API。所有组件都封装了 ZeroMQ 套接字生命周期,并对它们内部管理线程,从而使与这些组件的交互无锁和线程安全。 你需要将此依赖项包含在你的项目中:

  • Maven

  • Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-zeromq</artifactId>
    <version>{project-version}</version>
</dependency>
compile "org.springframework.integration:spring-integration-zeromq:{project-version}"

ZeroMQ Proxy

ZeroMqProxy 是内置 ZMQ.proxy() function 的 Spring 友好封装。它封装了套接字生命周期和线程管理。此代理程序的客户端仍然可以使用标准的 ZeroMQ 套接字连接和交互 API。除了标准 ZContext 之外,它还需要众所周知的 ZeroMQ 代理模式之一:SUB/PUB、PULL/PUSH 或 ROUTER/DEALER。这样就为代理程序的前端和后端使用了一对适当的 ZeroMQ 套接字类型。有关详细信息,请参阅 ZeroMqProxy.Type

ZeroMqProxy 实现 SmartLifecycle 来创建、绑定和配置套接字,以及在来自 Executor(如果有的话)的专用线程中启动 ZMQ.proxy()。前端和后端套接字的绑定通过 tcp:// 协议在所有可用的网络接口上进行,并使用提供的端口。否则,它们会绑定到稍后可以通过相应的 getFrontendPort()getBackendPort() API 方法获得的随机端口。

控制套接字公开为 SocketType.PAIR,在 "inproc://" + beanName + ".control" 地址上进行线程间传输;可以通过 getControlAddress() 获得它。它应当与来自另一个 SocketType.PAIR 套接字的相同应用程序一起使用,以发送 ZMQ.PROXY_TERMINATEZMQ.PROXY_PAUSE 和/或 ZMQ.PROXY_RESUME 命令。当调用 stop() 来终止 ZMQ.proxy() 循环并正常关闭所有绑定的套接字时,ZeroMqProxy 会执行 ZMQ.PROXY_TERMINATE 命令。

setExposeCaptureSocket(boolean) 选项会导致此组件使用 SocketType.PUB 绑定一个额外的线程间套接字,以便捕获和发布前端和后端套接字之间的所有通信,正如 ZMQ.proxy() 实现所述。此套接字绑定到 "inproc://" + beanName + ".capture" 地址,并且不要求任何特定订阅进行筛选。

可以通过其他属性,如读/写超时或安全性来定制前端和后端套接字。这种定制可分别通过 setFrontendSocketConfigurer(Consumer<ZMQ.Socket>)setBackendSocketConfigurer(Consumer<ZMQ.Socket>) 回调获得。

ZeroMqProxy 可作为以下的简单 Bean 提供:

@Bean
ZeroMqProxy zeroMqProxy() {
    ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT, ZeroMqProxy.Type.SUB_PUB);
    proxy.setExposeCaptureSocket(true);
    proxy.setFrontendPort(6001);
    proxy.setBackendPort(6002);
    return proxy;
}

所有客户端节点都应通过 tcp:// 连接到此代理的主机,并使用它们各自感兴趣的端口。

ZeroMQ Message Channel

ZeroMqChannel 是一个 SubscribableChannel,它使用一对 ZeroMQ 套接字来连接发布者和订阅者,以进行消息传递交互。它可以在 PUB/SUB 模式下工作(默认为 PUSH/PULL);它还用作本地线程间通道(使用 PAIR 套接字) - 在这种情况下,不提供 connectUrl。在分布式模式下,它必须连接到外部管理的 ZeroMQ 代理,在那里它可以与连接到同一代理的其他类似通道交换消息。连接 URL 选项是一个标准的 ZeroMQ 连接字符串,其中包含协议和主机,以及冒号分隔的用作 ZeroMQ 代理前端和后端套接字的一对端口。为了方便,如果在与代理相同的应用程序中配置通道,则可以使用 ZeroMqProxy 实例(而不是连接字符串)来提供通道。

发送和接收套接字在其各自的专用线程中进行管理,这使得此通道对并行处理很友好。这样,我们可以从不同的线程向/从 ZeroMqChannel 发布和使用数据,而无需同步。

默认情况下,ZeroMqChannel 使用 EmbeddedJsonHeadersMessageMapper 来使用 Jackson JSON 处理程序从 byte[] (de)序列化 Message(包括头信息)。此逻辑可以通过 `setMessageMapper(BytesMessageMapper)`进行配置。

通过各个 setSendSocketConfigurer(Consumer<ZMQ.Socket>)setSubscribeSocketConfigurer(Consumer<ZMQ.Socket>) 回调,可以为任何选项(读/写超时、安全性等)自定义发送和接收套接字。

ZeroMqChannel 的内部逻辑基于响应式流,通过项目反应器 FluxMono 操作符。这提供了更简单的线程控制,并允许对通道进行无锁并发发布和消费。本地 PUB/SUB 逻辑实现为 Flux.publish() 操作符,以允许该通道的所有本地订阅者接收相同的已发布消息,作为 PUB 套接字的分布式订阅者。

以下是一个 ZeroMqChannel 配置的简单示例:

@Bean
ZeroMqChannel zeroMqPubSubChannel(ZContext context) {
    ZeroMqChannel channel = new ZeroMqChannel(context, true);
    channel.setConnectUrl("tcp://localhost:6001:6002");
    channel.setConsumeDelay(Duration.ofMillis(100));
    return channel;
}

ZeroMQ Inbound Channel Adapter

ZeroMqMessageProducer 是一种带有响应式语义的 MessageProducerSupport 实现。它以非阻塞方式持续读取 ZeroMQ 套接字中的数据,并将消息发布到无限 Flux 中,该 FluxFluxMessageChannel 订阅,或者如果输出通道不是响应式,则在 start() 方法中明确订阅。当套接字上没有收到数据时,在下次读取尝试之前应用 consumeDelay(默认为 1 秒)。

ZeroMqMessageProducer 仅支持 SocketType.PAIRSocketType.PULLSocketType.SUB。此组件可以连接到远程套接字或使用提供的或随机端口绑定到 TCP 协议。可以在此组件启动后通过 getBoundPort() 获取实际端口,并且 ZeroMQ 套接字已绑定。套接字选项(例如安全性或写超时)可以通过 setSocketConfigurer(Consumer<ZMQ.Socket> socketConfigurer) 回调进行配置。

如果 receiveRaw 选项设置为 true,则从套接字中获取的 ZMsg 会按原样发送到生成消息的有效负载中:由下游流程解析并转换 ZMsg。否则,InboundMessageMapper 用于将获取的数据转换为 Message。如果接收到的 ZMsg 是多帧的,则第一帧将被视为 ZeroMQ 消息发布到的 ZeroMqHeaders.TOPIC 头。

对于 SocketType.SUBZeroMqMessageProducer 使用提供的 topics 选项进行订阅;默认为订阅所有。可以通过 subscribeToTopics()unsubscribeFromTopics() @ManagedOperation 在运行时调整订阅。

以下是 ZeroMqMessageProducer 配置的示例:

@Bean
ZeroMqMessageProducer zeroMqMessageProducer(ZContext context, MessageChannel outputChannel) {
    ZeroMqMessageProducer messageProducer = new ZeroMqMessageProducer(context, SocketType.SUB);
    messageProducer.setOutputChannel(outputChannel);
    messageProducer.setTopics("some");
    messageProducer.setReceiveRaw(true);
    messageProducer.setBindPort(7070);
    messageProducer.setConsumeDelay(Duration.ofMillis(100));
    return messageProducer;
}

ZeroMQ Outbound Channel Adapter

ZeroMqMessageHandler 是一种 ReactiveMessageHandler 实现,用于将发布消息生成到 ZeroMQ 套接字中。只支持 SocketType.PAIRSocketType.PUSHSocketType.PUBZeroMqMessageHandler 只支持连接 ZeroMQ 套接字;不支持绑定。当使用 SocketType.PUB 时,将根据请求消息评估 topicExpression,以在 ZeroMQ 消息中注入主题帧(如果它不为 null)。订阅方 (SocketType.SUB) 必须先接收主题帧,然后才能解析实际的数据。当请求消息的有效负载为 ZMsg 时,不执行转换或主题提取:ZMsg 被原样发送到套接字中,并且不会将其销毁以备后续可能的重复使用。否则,使用 OutboundMessageMapper<byte[]> 将请求消息(或仅其有效负载)转换为 ZeroMQ 帧以发布。默认情况下,使用带有 ConfigurableCompositeMessageConverterConvertingBytesMessageMapper。套接字选项(例如安全性或写超时)可以通过 setSocketConfigurer(Consumer<ZMQ.Socket> socketConfigurer) 回调进行配置。

以下是 ZeroMqMessageHandler 配置的示例:

@Bean
@ServiceActivator(inputChannel = "zeroMqPublisherChannel")
ZeroMqMessageHandler zeroMqMessageHandler(ZContext context) {
    ZeroMqMessageHandler messageHandler =
                  new ZeroMqMessageHandler(context, "tcp://localhost:6060", SocketType.PUB);
    messageHandler.setTopicExpression(
                  new FunctionExpression<Message<?>>((message) -> message.getHeaders().get("topic")));
    messageHandler.setMessageMapper(new EmbeddedJsonHeadersMessageMapper());
}

ZeroMQ Java DSL Support

spring-integration-zeromq 通过 ZeroMq 工厂和上述组件的 IntegrationComponentSpec 实现提供了一个便捷的 Java DSL 流畅 API。

以下是 ZeroMqChannel 的 Java DSL 示例:

.channel(ZeroMq.zeroMqChannel(this.context)
            .connectUrl("tcp://localhost:6001:6002")
            .consumeDelay(Duration.ofMillis(100)))
}

ZeroMQ Java DSL 的入站通道适配器是:

IntegrationFlow.from(
            ZeroMq.inboundChannelAdapter(this.context, SocketType.SUB)
                        .connectUrl("tcp://localhost:9000")
                        .topics("someTopic")
                        .receiveRaw(true)
                        .consumeDelay(Duration.ofMillis(100)))
}

ZeroMQ Java DSL 的出站通道适配器是:

.handle(ZeroMq.outboundChannelAdapter(this.context, "tcp://localhost:9001", SocketType.PUB)
                  .topicFunction(message -> message.getHeaders().get("myTopic")))
}