spring integration-UDP和TCP模块

水深无声 2023-07-09 15:27 14阅读 0赞

## 概述 ##

TBD

## UDP适配器 ##

TBD

## TCP连接工厂 ##

**概述**  
对于TCP来说，底层的连接配置是由连接工厂提供。提供两种类型的连接工厂：客户端连接工厂和服务端连接工厂。客户端连接工厂建立出站连接。服务端连接工厂监听入站连接。

一个输出通道适配器使用一个客户端连接工厂，但是也可以提供一个客户端连接工厂给输入通道适配器。这个适配器接收输出通道适配器建立的连接发来的消息。（注：输出通道适配器和输入通道适配器共用一个客户端连接工厂）

一个输入通道适配器或网关使用一个服务端连接工厂（没有一个连接工厂无法工作）。可以提供一个服务端连接工厂给输出通道适配器。可以使用这个适配器发送应答给同连接发来的消息。

> 只有当应答包含ip\_connectionId消息头（被连接工厂插入的）时，应答消息才会被路由到连接上。  
> 当在输入和输出适配器共享连接工厂时，会涉及到消息的关联程度。这种共享允许tcp双向异步通信。默认情况下，只有消息负载信息在TCP上传输。这样的话，任何消息关联必须在下游的组件比如聚合器或其他端点上进行。版本3.0后支持给定的消息头。更多信息，请参考TCP消息关联。

只能给一个一种类型的适配器最多一个连接工厂。

Spring Integration提供使用java.net.Socket和java.nio.channel.SocketChannel的连接工厂。

下面的例子展示一个简单的使用java.net.Socket连接的连接工厂：

<int-ip:tcp-connection-factory id="server"
      type="server"
    
      port="1234"/>

下面是使用java.nio.channel.SocketChannel连接的连接工厂：

<int-ip:tcp-connection-factory id="server"
      type="server"
      port="1234"
      using-nio="true"/>

> 从版本4.2开始，如果一个服务器配置为监听随机端口（设置端口为0），可以通过getPort()方法获取实际绑定的端口。另外，getServerSocketAddress()方法可以获取SocketAddress。参考TcpServerConnectionFactory接口的doc文档获取更多信息。

<int-ip:tcp-connection-factory id="client"
      type="client"
      host="localhost"
      port="1234"
      single-use="true"
      so-timeout="10000"/>

上面的代码块展示了使用java.net.Socket的客户端连接工厂，并且为每个消息创建一个连接。

<int-ip:tcp-connection-factory id="client"
      type="client"
      host="localhost"
      port="1234"
      single-use="true"
      so-timeout="10000"
      using-nio=true/>

从版本5.2开始，客户端连接支持属性connectTimeout，单位是秒，默认60。

**消息划分（序列化器和反序列化器）**  
TCP是一个流协议。这意味着一些结构必须提供给传输的数据以便于接收者可以把数据划分为独立的消息。可以给连接工厂配置序列化器和反序列器来实现TCP上消息负载和数据位之间的转换。实现这个可以给入站消息和出战消息提供一个反序列化器和序列化器，同样地，Spring Integration提供了一些标准的序列化器和反序列化器。

ByteArrayCrlfSerializer\*转换一个字节数组到以返回和换行（\\r\\n）字符结尾的字节流。这是默认的序列化器（和反序列化器）可以用来（比如）telnet客户端。

ByteArraySingleTerminatorSerializer\*转换一个字节数组为一个以单个结束字符（默认0x00）的字节流。

ByteArrayLfSerializer\*转换一个字节数组为一个以单个返回字符（0x0a）结尾的字节流。

ByteArrayStxEtxSerializer\*转换一个字节数组为一个以STX（0x02）开头和ETX（0x03）结尾的字节流。

ByteArrayLengthHeaderSerializer转换一个字节数组到一个以网络字节序（大端编码）的二进制长度开头的字节流。【准备中】

ByteArrayRawSerializer\*转换一个字节数组到一个没有添加任何额外划分信息的字节流。使用序列化器时，以客户端关闭socket作为消息的结束标志。【准备中】

> 版本4.2.2之前，使用非阻塞I/O时，序列化器把超时作为消息的结束，把超时之前的数据作为消息发出。这是不可靠的，不应该被用来界定消息。现在这种情况被认为是异常。并不建议这么做，但是可以通过设置treatTimeoutAsEndOfMessage为true来打开这种行为。

每个序列化器和反序列化器都是AbstractByteArraySerializer子类，同时实现org.springframework.core.serializer.Serializer和org.springframework.core.serializer.Deserializer。为了向后兼容，任何使用AbstractByteArraySerializer子类序列化的连接接收一个首先转成字节数组的String类型。每个序列化器和反序列化器特定格式的输入流转成字节数组负载。

为了避免由于客户端的不当行为（没有指定配置的序列化器）导致内存耗尽，这些序列化器限制一个最大的消息大小。如果入站消息超过这个大小，一个异常抛出。默认的最大消息大小是2048字节。可以设置maxMessageSize属性来修改这个大小。如果想要默认的序列化器和反序列化器增加这个最大消息大小，必须显示指定连接工厂使用的序列化对象的maxMessageSize属性。

本章节上面标\*的类使用一个临时的缓冲区并把解码数据复制到正确大小的不可变缓冲中。从版本4.3开始，可以指定一个poolSize属性来使得原始缓冲区复用，而不是为每个消息都分配和清除缓冲，这是默认的行为。设置为负数使得缓冲区没有边界。如果缓冲区有边界，可以设置poolWaitTimeout（单位毫秒）属性，超过这个时间没有数据到来会抛出异常。默认是无穷大。这个异常会导致socket关闭。

如果想让自定义的解码器具有这种行为，可以继承AbstractPooledBufferByteArraySerializer（而不是它的父类，AbstractByteArraySerializer）并且实现doDeserialize()而不是deserialize()。缓冲区自动返回给缓冲池。AbstractPooledBufferByteArraySerializer同样提供了一个方便的工具方法：copyToSizedArray()。

版本5.0添加ByteArrayElasticRawDeserializer。这是一个和上面提到的ByteArrayRawSerializer类似的解码器，除了它不需要设置maxMessageSize之外。内部使用ByteArrayOutputStream使得缓冲区按需增长。客户端必须关闭socket来结束消息。

MapJsonSerializer使用Jackson的ObjectMapper来转换Map和JSON。可以使用序列化器联合MessageConvertingTcpMessageMapper和MapMessageConverter以JSON格式传输消息头和负载。

> Jackson的ObjectMapper无法界定流中的消息。这样，MapJsonSerializer需要代理另外一个序列化器或反序列化器来处理消息划分。默认使用ByteArrayLfSerizlizer，通道中流的格式为jsonLF，当然也可以使用其他配置。

最后一个标准序列化器是 org.springframework.core.serializer.DefaultSerializer，使用Java序列化技术实现可序列化对象的转化。org.springframework.core.serializer.DefaultDeserializer用来反序列化流中包含的序列化对象。

如果不想用默认的序列化器和反序列化器（ByteArrayCrLfSerializer），必须设置连接工厂的serializer和deserializer属性。

看下面例子：

一个服务端连接工厂，使用java.net.Socket连接，并使用Java序列化技术序列化通道中的数据。

查看连接工厂的所有可用属性，查看本节最后的附录。

默认情况下，消息入站时，反DNS服务把IP地址转成主机名，并存在消息头中。在没有配置DNS的环境中，会导致连接延迟。可以设置lookup-host属性来重载默认行为。

> 可以修改sockets和socket工厂的属性。参考SSL/TLS获取更多信息。如前面提到，这些修改决定SSL是否生效。

**自定义序列化器和反序列化器**

如果标准的反序列化器不能支持数据的格式，可以自定义实现；也可以实现一个自定义的序列化器。

为实现序列化器和反序列化器对，实现org.springframework.core.serializer.Deserializer和org.springframework.core.serializer.Serializer两个接口。

当反序列化器检测到一个关闭输入流的信号，必会抛出一个SoftEndOfStreamException异常；这是一个框架表示一个正常关闭的信号。如果流在解码过程中关闭，会抛出其他的异常。

版本5.2开始，SoftEndOfStreamException是一个RuntimeException而不是IOException。

**TCP缓存客户端连接工厂**

前面讲到，TCP sockets可以‘“single-use”（单个请求或应答）或共享。共享socket在出站网关高容量环境中表现一般，因为socket一次只能处理一个请求或应答。

为了提高性能，使用通道适配器而不是网关，但是需要程序层面的消息关联。查看TCP消息关联查看更多信息。

版本2.2引入缓存客户端连接工厂，持有一个共享socket池，让网关能够处理一个共享连接池中的多个并发请求。

**TCP容错客户端连接工厂**

配置TCP连接工厂支持多个服务器的容错。发送一个消息时，工厂遍历配置的所有工厂直到消息能被发送或者没有连接可用。初始化时，使用第一个工厂。如果一个连接频繁失败，下一个工厂变成当前工厂。下面展示如何配置一个容错客户端连接工厂。

> 配置容错连接工厂时，singleUse属性必须在工厂本身和工厂列表中保持一致。

**TCP线程归属连接工厂**

版本5.0引入这个连接工厂。该工厂绑定一个连接到调用线程，每次该线程发送一个消息时，重用相同的连接发送。这种状况一直持续到连接关闭（被服务器或者网络关闭）或线程调用releaseConnection()。连接是由另外一个工厂提供的，该工厂必须配置非共享（single-use）连接，以便每个线程拿到一个连接。

下面例子展示了TCP线程归属连接工厂配置：

@Bean
    public TcpNetClientConnectionFactory cf() {
      TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory( "localhost",
      Integer.parseInt(System.getProperty(PORT)));
      cf.setSingleUse(true);
      return cf; }
    @Bean
    public ThreadAffinityClientConnectionFactory tacf() {
      return new ThreadAffinityClientConnectionFactory(cf()); }
    @Bean
    @ServiceActivator(inputChannel = "out")
    public TcpOutboundGateway outGate() {
      TcpOutboundGateway outGate = new TcpOutboundGateway();
      outGate.setConnectionFactory(tacf());
      outGate.setReplyChannelName("toString");
      return outGate; }

## TCP连接拦截器 ##

配置连接工厂的TcpConnectionInterceptorFactoryChain。使用拦截器添加连接的行为，比如协商，加密和其他。框架没有提供拦截器，可以查看代码的仓库的InterceptedSharedConnectionTests参考。

测试用例的HelloWorldInterceptor有以下功能：

拦截器首先配置到客户端连接工厂。第一个消息通过拦截器连接发送时，拦截器发送‘Hello’并期望应答‘World!’。这个过程发生后，协商完成，消息开始发送。后续的消息使用相同的连接发送而不需要额外的协商。

配置服务端连接工厂时，拦截器请求第一个‘Hello’消息并返回‘world!’。否则抛出异常导致连接关闭。

所有TCPConnection方法都被拦截。拦截器工厂为连接创建拦截器。如果拦截器有状态，工厂为每个连接创建一个新的拦截器。如果没有状态，则相同的拦截器包装每个连接。拦截器工厂可以通过设置连接工厂的interceptor-factory-chain属性加入到拦截器工厂链中。拦截器必须继承TcpConnectionInterceptorSupport，工厂必须实现TcpConnectionInterceptorFactory接口。TcpConnectionInterceptorSupport有空方法。继承该类时，只需实现想拦截的方法即可。

下面例子演示了如何配置连接拦截器工厂链：

## TCP连接事件 ##

版本3.0开始，TcpConnection的改变以TcpConnectionEvent的形式通知。TcpConnectionEvent是ApplicationEvent子类，所以可以被任何在ApplicationContext中定义的ApplicationListener接收-比如[事件入站通道适配器][Link 1]。

TCPConnectionEvents有以下属性：

*  connectionId：连接标识符，用在消息头指定发送数据的连接。
 *  connectionFactoryName：连接所属的连接工厂名称。
 *  throwable：异常（仅TcpConnectionExceptionEvent事件）。
 *  source：TcpConnection。比如，使用getHostAddress()查看远程IP地址。

另外，版本4.0之后，前面讨论的标准反序列化器在解码数据流发生错误时抛出TcpDeserializationExceptionEvent实例。这些事件包含异常，缓冲区和异常发生时的偏移量。程序可以使用一般的ApplicationListener或ApplicationEventListeningMessageProducer来捕获这些事件分析问题。

版本4.0.7和4.1.3开始，无论服务端socket（比如端口占用的BindException）发生什么错误都会发布TcpConnectionServerExceptionEvent事件。这些事件包含连接工厂和原因的引用。

版本4.2开始，无论一个端点（入站网关或出站通道适配器）收到端点后，由于ip\_connection消息头无效，导致消息无法路由到连接，会发布TcpConnectionFailedCorrelationEvent。当发送消息应答迟到（发送线程超时）时，出站网关也会发布这个事件。这个事件包含连接ID以及包含错误的cause属性。

版本4.3开始，TcpConnectionServerListeningEvent会在服务端连接工厂开启时发布。这在设置端口为0，由操作系统选择端口时非常有用。当等待服务绑定到端口然后开启一些进程时，监听这个事件比轮询isListening()要好得多。

> 为了避免监听线程接收连接请求，这个消息在一个单独的线程发布。

版本4.3.2开始，TcpConnectionFailedEvent会在无法创建客户端连接时发布。事件的源是连接工厂，可以用来查看是哪个主机和端口无法建立连接。

## TCP适配器 ##

TCP入站和出站通道适配器使用上面提到的连接工厂。这些适配器有两个重要的属性，connection-factory和channel。connection-factory用来指定适配器用来管理连接的连接工厂。channel属性指定出站适配器消息到达的通道和入站适配器放置消息的通道。入站和出站适配器可以共享一个连接工厂，但服务端连接工厂总是被入站适配器所拥有。客户端连接工厂总是被出站适配器所拥有。每种类型的适配器只有一个能引用到连接工厂。下面展示了如何定义客户端和服务端TCP连接工厂：

上面的配置中，达到input通道的消息在client连接工厂创建的连接上序列化，在服务器上接收，并放置在loop通道上。因为loop是outboundServer的入站通道，消息在同一连接上回环，被inboundClient接收，并放置到replies通道。管道上使用java序列化。

通常情况下，入站适配器使用type="server"连接工厂，监听入站连接请求。有些情况下，需要建立相反的连接，入站适配器连接到外部服务器并等待连接的入站消息。

这种拓扑可以设置入站适配器的client-mode="true"来支持。这种情况下，连接工厂必须是client类型并且single-use必须设置为false。

两个额外的属性支持这种结构。retry-interval（毫秒）指定连接失败重新尝试连接的频率。scheduler提供一个TaskScheduler来调度重连任务并测试连接是否依然活跃。

如果不指定调度器，框架使用默认的taskScheduler。

对于出站适配器，连接通常在第一个消息发送时建立。出站适配器设置client-mode="true"使得连接在适配器启动时就被建立。默认情况下，适配器自动启动。同样，连接工厂必须是client模式并且single-use=“false”。retry-interval和scheduler依然被支持。如果连接出错，重连被调度器建立或发送下一个消息时建立。

对于入站和出站，如果适配器启动，可以通过发送一个`<control-bus />`命令：@adapter\_id.retryConnections()，来强制向适配器建立一个连接。然后通过@adapter\_id.isClientModeConnected()来检查当前状态。

## TCP网关 ##

入站TCP网关TcpInboundGateway和出战TCP网关TcpOutboundGateway分别使用服务端连接工厂和客户端连接工厂。每个连接每次处理单个请求或应答。

入站网关，在用入站负载构造了消息并发送到requestChannel之后，等待应答并发送应答消息的负载到连接上。

> 对于入站网关，必须保留或增加ip\_connectionId消息头，用来关联消息和连接。由网关创建的消息自动设置了这个消息头。如果应答构造了新消息，需要设置这个消息头。消息头的值可以从入站消息中获取。

类似入站适配器，入站网关一般使用type="server"的连接工厂，监听入站连接请求。有些情况下，会建立反向的连接，比如入站网关连接外部服务器然后等待发送到连接的应答。

这个拓扑可以设置入站网关client-mode="true"来支持。这种情况下，连接工厂必须是client类型并且single-use置为false。

两个额外的属性支持这种结构。retry-interval（毫秒）指定连接失败后重连的频率。scheduler指定TaskScheduler调度重连任务和测试连接是否活跃。

一旦网关启动，可以通过发送一个`<control-bus />`命令：@adapter\_id.retryConnections()，来强制向适配器建立一个连接。然后通过@adapter\_id.isClientModeConnected()来检查当前状态。

出站网关在连接上发送一个消息后，等待应答，构造一个应答消息，放入应答通道内。和所有连接的通讯时单线程的。一次只能有一个消息被处理。如果另外一个线程在应答到达之前发送消息，该线程会一直阻塞到前一个请求完成。然而，客户端连接工厂配置为single-use连接类型，每个新的请求有单独的连接并且会被马上处理。下面的例子配置了一个入站TCP网关：

<int-ip:tcp-inbound-gateway id="inGateway"
      request-channel="tcpChannel"
      reply-channel="replyChannel"
      connection-factory="cfServer"
      reply-timeout="10000"/>

如果连接工厂使用默认的序列化器和反序列化器，消息是以\\r\\n分割的数据并且网关能够被简单的客户端比如telnet访问。

下面例子展示了出站TCP网关：

<int-ip:tcp-outbound-gateway id="outGateway"
      request-channel="tcpChannel"
      reply-channel="replyChannel"
      connection-factory="cfClient"
      request-timeout="10000"
      remote-timeout="10000"/>

client-mode属性对出站网关并不可用。

版本5.2开始，出站网关可以配置closeStreamAfterSend属性。如果连接工厂配置了single-use（为每个请求/应答建立新连接），网关会关闭输出流；这会发送EOF信号给服务器。这样服务器通过EOF而不是流的其他分割符来判断消息的结束，并且保持连接打开接收应答。

## TCP消息关联 ##

IP端点的目的是为了给系统而不是Spring Integration程序提供通讯的。基于这个原因，默认只有消息负载被发送和接收。3.0版本开始，可使用JSON，Java序列化和自定义序列化器和反序列化器来转移消息头。参见转移消息头获取更多信息。框架（除了网关）不提供消息关联和协商服务端的通道适配器。文档的后面部分，会讨论程序层面的各种关联技术。大部分情况下，即使消息负载包含一些关联数据（比如有序编号），还是需要指定程序级别的消息关联。

**网关**

网关自动关联消息。然而，需要在吞吐量较小的程序中使用出站网关。配置了连接工厂使用单个共享连接处理消息对（single-use=false）时，一次只有处理一个消息。新的消息必须等待上个消息的应答到达。如果连接工厂配置为每个新的消息新建一个连接时（single-use=true），这个限制不成立。这样的配置比共享连接带来更高的吞吐量，但是也伴随着打开和关闭连接带来的开销。

所有，对于高吞吐量的消息，考虑使用协调一对通道适配器。然而，这样做的话，需要提供协调逻辑。

**协调出站和入站通道适配器**

为了实现高吞吐量（规避网关的缺点），需协调一对出站和入站通道适配器。可以使用协调适配器（服务端或客户端）来进行异步通讯（而不是请求-应答语义）。在服务端，适配器自动处理消息关联，因为入站适配器添加一个消息头允许出站适配器决定用那个连接发送应答消息。

> 在服务端，必须配置ip\_connectionId消息头，因为这个消息头关联消息和连接。入站适配器创建的消息自动添加这个消息头。如果想构造其他的消息发送，需要设置这个消息头。可以从入站消息拿到这个消息头。

在客户端，如果需要的话，程序必须提供自定义的关联逻辑。有一系列的方法。

如果消息负载包含关联数据（比如事务ID或有序编号），那么没有必要从原始的出站消息获取任何信息（比如应答通道适配器），这种一种简单的关联，务必在程序层面进行。

如果消息负载包含关联数据（比如事务ID或有序编号），需要从原始出站消息保留一些信息（比如应答通道头），可以保留一份原始出站数据的副本（可能在一个发布-订阅通道中使用的），并使用聚合器重组必要的数据。

除了上面两种场景，如果负载没有关联数据，可以在上游提供一个转换器往负载注入关联数据。转换器转换原来的负载到一个包含原来负载和消息头子集的新消息。当然消息头的活跃对象（比如应答通道）不能包含到转换后的负载上。

如果使用这种策略，需保证连接工厂有合适的序列化器-反序列化器组（比如DefaultSerializer和DefaultDeserializer）来处理负载。前面提到的ByteArray\*Serializer，包括默认的ByteArrayCrLfSerializer，不支持这样的负载，除非要转换的负载是一个字符串或byte\[\]。

> 版本2.2之前，协作通道适配器使用客户端连接工厂，so-timeout属性默认设置为默认的应答超时时间（10s）。这意味着，如果一段时间没有数据到达入站适配器，socket就会关闭。  
> 这种默认行为再可靠的异步环境中并不合适，所以现在默认设置为无穷大超时时长。可以通过设置客户端连接工厂的so-timeout属性为10000毫秒来重新配置前面的默认行为。

**转移消息头**

TCP是一种流协议。Serializer和Deserializers划分流中的消息。版本3.0之前，只有消息负载（String或byte\[\]）可以通过TCP传输。版本3.0开始，特定的消息头也能像负载一样传输。然而，“活跃”对象，比如replyChannel消息头，不能被序列化。

发送消息头信息到TCP上需要额外的配置。

第一步是使用mapper属性为ConnectionFactory提供MessageConvertingTcpMessageMapper对象。该对象代理任何MessageConverter实现来转换消息到对象和对象到消息，这些对象通过配置的serializer和deserializer来序列化和反序列化。

Spring Integration提供MapMessageConverter，使得消息头列表和负载加入到一个Map对象。这个Map有两个入口：payload和headers。Map的headers入口包含这些消息头。

第二步是提供序列化器和反序列化器实现Map和线路上数据的转换。可以是自定义的Serializer和Deserializer，这个序列化是必须的，如果远程系统不是Spring Integration程序的话。

Spring Integration提供MapJsonSerializer来转换Map和JSON。使用Spring Integration JsonObjectMapper。如果必要的话提供一个自定义的JsonObjectMapper。默认情况下，序列化器在对象之间加入一个换行（0x0a）符。

> JsonObjectMapper使用类路径上任何版本的Jackson。

也可以用DefaultSerializer和DefaultDeserializer来标准Java序列化Map。

下面的例子展示了一个用JSON传输correlationId，sequenceNumber和sequenceSize消息头的连接工厂的配置：

<int-ip:tcp-connection-factory id="client"
      type="client"
      host="localhost"
      port="12345"
      mapper="mapper"
      serializer="jsonSerializer"
      deserializer="jsonSerializer"/>
    <bean id="mapper"
      class="o.sf.integration.ip.tcp.connection.MessageConvertingTcpMessageMapper
    ">
      <constructor-arg name="messageConverter">
      <bean class="o.sf.integration.support.converter.MapMessageConverter">
      <property name="headerNames">
      <list>
      <value>correlationId</value>
      <value>sequenceNumber</value>
      <value>sequenceSize</value>
      </list>
      </property>
      </bean>
      </constructor-arg>
    </bean>
    <bean id="jsonSerializer" class= "o.sf.integration.ip.tcp.serializer.MapJsonSerializer" />

有‘something’负载的消息通过上面的配置发送后，在线路上呈现为：

{"headers":{"correlationId":"things","sequenceSize":5,"sequenceNumber":1},"payload
    ":"something"}

## 非阻塞I/O ##

使用NIO（参见using-nio）避免给每个socket创建一个线程。对于小数量的socket，不用NIO，使用异步切换（比如QueueChannel）比NIO的性能有过之而无不及。

对于大数量的连接要考虑使用NIO。然而，使用NIO有一些难以预料的后果。线程池（task executor中）在所有的socket间共享。每个收到的消息被线程池的一个线程发送到通道的任务被认为是独立的单元。到达socket上两个有序的消息可能被不同的线程处理。这意味着发送到通道的消息的顺序是不确定的。到达socket的消息顺序不能被严格保证。

对某些程序来说，这并不是个问题。但对另外一些，确是问题。如果需要严格的顺序，考虑设置using-nio为false并使用异步切换。

另外，可以插入一个重排器到入站端点的下游并返回消息的正确顺序。设置连接工厂的apply-sequence为true，到达TCP连接的消息就会设置sequenceNumber和correlationId。重排器使用这些消息头使消息有正确的顺序。

> 版本5.1.4开始，给现有连接的基础上创建新连接读数据设置优先级。这个，一般来说，影响较小，除非有高频的新连接创建。如果想赋予读取数据优先级，设置TcpNioServerConnectionFactory的multiAccept属性为false。

**池大小**

池大小属性不再使用。先前，当task-executor未指定时，这个属性用来指定默认线程池的大小。也用来设置服务器连接的存储。第一个功能已不在需要（看下一章节）。第二个功能被backlog属性替代。

先前，NIO使用固定线程池大小（默认）的任务执行器，可以会死锁并导致处理过程停止。当缓冲已满，一个线程尝试从socket读取数据并把数据加到缓冲，没有多余的线程用来分配缓冲的空间时，这个问题产生了。这只在池很小时才会发生，并且在极端的条件才有可能。版本2.2开始，有两处修改来解决这个问题。首先，默认任务执行器变为缓存线程池执行器。其次，添加死锁检测逻辑，如果线程饥饿发生，会抛出异常释放死锁的资源而不是一直死锁。

> 注意默认任务执行器是没有边界的，如果消息处理时间过长，并且入站消息频率极高，内存不足的问题就发生了。如果是这种类型的程序，需要用一个合适大小的任务执行器。参见下一章节。

**带CALLER\_RUNS政策的线程池任务执行器**

使用带CallerRunsPolicy的固定大小的线程池时，有重要的几点需要考虑，并且队列容量小。

使用固定大小线程池时，下列情况并不适用：

使用NIO连接时，有三个不同的任务类型。I/O选择器在一个专门的线程上运行（检测事件，接收新连接，使用任务执行器分发I/O读操作到其他线程。当I/O读取线程读取消息时，会切换到另一个线程收集消息。大消息会读取几次才完成。“收集器”等待数据时会一直阻塞。当一个新的读事件发生时，读取器判断socket是否已有收集器，如果没有，运行一个新的。当读取过程结束后，读取器线程返回给线程池。

当池用完后，设置了CALLER\_RUNS拒绝策略，并且任务队列已满，这会导致死锁。当线程池已经用完，并且队列没有空间时，IO选择器线程收到一个OP\_READ事件，并使用执行器分发读取操作。因为队列是满的，所以选择器线程自身开启读过程。选择器在开始读之前，检测到这个socket没有收集器，便会开启一个收集器（注：一般是个循环）。如上，队列是满的，选择器线程变成了收集器。这个收集器现在因为等待不会到来的数据而阻塞。连接工厂现处于死锁状态，因为选择器线程不能处理新的事件。

为了避免死锁，必须避免选择器（或读取器）线程执行收集任务。可以使用其他的线程池来处理IO和收集操作。

框架提供CompositeExecutor，允许配置两个独立的执行器：一个执行IO操作，一个收集消息。这种情况下，IO操作永远不会变成收集器线程，从而死锁不会发生。

另外，任务执行器可以配置使用AbortPolicy（ABORT使用`<task>`）。当一个I/O任务不能完成时，等待一会然后持续重试直到任务能够完成并分配了一个收集器。默认情况下，延迟是100ms，可以设置连接工厂的readDelay属性来修改这个值（xml配置是read-delay）。

下面三个例子展示如何配置组合执行器：

@Bean
    private CompositeExecutor compositeExecutor() {
      ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
      ioExec.setCorePoolSize(4);
      ioExec.setMaxPoolSize(10);
      ioExec.setQueueCapacity(0);
      ioExec.setThreadNamePrefix("io-");
      ioExec.setRejectedExecutionHandler(new AbortPolicy());
      ioExec.initialize();
      ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
      assemblerExec.setCorePoolSize(4);
      assemblerExec.setMaxPoolSize(10);
      assemblerExec.setQueueCapacity(0);
      assemblerExec.setThreadNamePrefix("assembler-");
      assemblerExec.setRejectedExecutionHandler(new AbortPolicy());
      assemblerExec.initialize();
      return new CompositeExecutor(ioExec, assemblerExec); }

## SSL/TLS支持 ##

[Link 1]: https://docs.spring.io/spring-integration/docs/5.2.3.RELEASE/reference/html/event.html#applicationevent