我的 Spring 集成场景是:
- 使用自定义协议发送数据的数十个生产者(大小和内容)
- 我必须解码这个自定义协议,然后处理结果。
所以我尝试了很多配置,目前最好的配置如下:
<bean id="serializer" class="com.MySerializerDeserializer" />
<task:executor id="myTaskExecutor" pool-size="5-300" queue-capacity="500000"/>
<int-ip:tcp-connection-factory id="serverTcpConFact"
type="server"
port="5566"
using-nio="true"
single-use="false"
so-timeout="5000"
task-executor="myTaskExecutor"
deserializer="serializer"
serializer="serializer"/>
<int-ip:tcp-inbound-channel-adapter id="tcpInboundAdapter"
channel="tcpInbound"
connection-factory="serverTcpConFact" />
<int:channel id="tcpInbound" />
<int:service-activator input-channel="tcpInbound"
ref="importService"
method="handler" />
<bean id="importService" class="com.MyImportService" />
序列化类是:
public class MySerializerDeserializer implements Serializer< MyMessage >, Deserializer< MyMessage > {
@Override
public MyMessage deserialize(InputStream inputStream) throws IOException {
DataInputStream dis = new DataInputStream(inputStream);
int size = dis.readInt();
byte[] b = new byte[size];
dis.read(b);
MyMessage s = new MyMessage ();
String value = new String(b);
s.setTest(value);
return s;
我使用此代码来测试服务器:
class SimpleThread extends Thread {
public void run() {
try {
String id = java.util.UUID.randomUUID().toString();
try (Socket echoSocket = new Socket("localhost", 5566)) {
DataOutputStream dos = new DataOutputStream(echoSocket.getOutputStream());
for (int i = 0; i < 500; i++) {
String s = id + " " + i;
dos.writeInt(s.length());
dos.write(s.getBytes());
dos.flush();
System.out.println(id + " - " + i + " - " + s.length());
}
}
当我使用单个线程执行此操作时,当我尝试执行多个线程时效果很好,例如:
for (int i = 0; i < 5; i++) {
new SimpleThread().start();
}
spring 集成服务器卡住了,我有以下警告:
WARN _[m [THREAD ID=myTaskExecutor-1] 2014-06-25 12:42:18 WARN org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory:566 - Timing out TcpNioConnection 127.0.0.1:56273:5566:4e3caf61-1101-4881-a1cf-1c31610b33f3
而且它不起作用,服务器无法接收消息。
With this error:
我哪里错了?
谢谢。
EDIT
我这样修改了线程轮询:
<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
<constructor-arg>
<bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="threadNamePrefix" value="io-" />
<property name="corePoolSize" value="4" />
<property name="maxPoolSize" value="8" />
<property name="queueCapacity" value="50000" />
<property name="rejectedExecutionHandler">
<bean class="org.springframework.integration.util.CallerBlocksPolicy">
<constructor-arg value="10000" />
</bean>
</property>
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="threadNamePrefix" value="assembler-" />
<property name="corePoolSize" value="4" />
<property name="maxPoolSize" value="10" />
<property name="queueCapacity" value="50000" />
<property name="rejectedExecutionHandler">
<bean class="org.springframework.integration.util.CallerBlocksPolicy">
<constructor-arg value="10000" />
</bean>
</property>
</bean>
</constructor-arg>
</bean>
服务器响应速度更快,但我仍然遇到问题
**编辑*****
进口服务是:
public class ImportService {
public void handler(MyMessage inp) {
System.out.println(Thread.currentThread().getName() + "******" + inp.getTest());
}