我的 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
而且它不起作用,服务器无法接收消息。
出现这个错误:
我哪里错了? 谢谢你。
编辑
我这样修改线程轮询:
<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());
}
最佳答案
我刚刚完全按照描述运行了您的测试(复制了您的代码)并且一切都按预期工作。 500 条消息传递给另一方。
我建议您打开 TRACE 级别的日志记录,并可能向反序列化器添加一些调试日志记录。
关于java - spring 集成超时客户端,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/24406645/
我真的很习惯使用Ruby编写以下代码:my_hash={}my_hash['test']=1Java中对应的数据结构是什么? 最佳答案 HashMapmap=newHashMap();map.put("test",1);我假设? 关于java-等价于Java中的RubyHash,我们在StackOverflow上找到一个类似的问题: https://stackoverflow.com/questions/22737685/
我正在使用Ruby2.1.1和Rails4.1.0.rc1。当执行railsc时,它被锁定了。使用Ctrl-C停止,我得到以下错误日志:~/.rvm/gems/ruby-2.1.1/gems/spring-1.1.2/lib/spring/client/run.rb:47:in`gets':Interruptfrom~/.rvm/gems/ruby-2.1.1/gems/spring-1.1.2/lib/spring/client/run.rb:47:in`verify_server_version'from~/.rvm/gems/ruby-2.1.1/gems/spring-1.1.
我在app/helpers/sessions_helper.rb中有一个帮助程序文件,其中包含一个方法my_preference,它返回当前登录用户的首选项。我想在集成测试中访问该方法。例如,这样我就可以在测试中使用getuser_path(my_preference)。在其他帖子中,我读到这可以通过在测试文件中包含requiresessions_helper来实现,但我仍然收到错误NameError:undefinedlocalvariableormethod'my_preference'.我做错了什么?require'test_helper'require'sessions_hel
有没有办法在这个简单的get方法中添加超时选项?我正在使用法拉第3.3。Faraday.get(url)四处寻找,我只能先发起连接后应用超时选项,然后应用超时选项。或者有什么简单的方法?这就是我现在正在做的:conn=Faraday.newresponse=conn.getdo|req|req.urlurlreq.options.timeout=2#2secondsend 最佳答案 试试这个:conn=Faraday.newdo|conn|conn.options.timeout=20endresponse=conn.get(url
我一直很高兴地使用DelayedJob习惯用法:foo.send_later(:bar)这会调用DelayedJob进程中对象foo的方法bar。我一直在使用DaemonSpawn在我的服务器上启动DelayedJob进程。但是...如果foo抛出异常,Hoptoad不会捕获它。这是任何这些包中的错误...还是我需要更改某些配置...或者我是否需要在DS或DJ中插入一些异常处理来调用Hoptoad通知程序?回应下面的第一条评论。classDelayedJobWorker 最佳答案 尝试monkeypatchingDelayed::W
我正在尝试使用boilerpipe来自JRuby。我看过guide从JRuby调用Java,并成功地将它与另一个Java包一起使用,但无法弄清楚为什么同样的东西不能用于boilerpipe。我正在尝试基本上从JRuby中执行与此Java等效的操作:URLurl=newURL("http://www.example.com/some-location/index.html");Stringtext=ArticleExtractor.INSTANCE.getText(url);在JRuby中试过这个:require'java'url=java.net.URL.new("http://www
我只想对我一直在思考的这个问题有其他意见,例如我有classuser_controller和classuserclassUserattr_accessor:name,:usernameendclassUserController//dosomethingaboutanythingaboutusersend问题是我的User类中是否应该有逻辑user=User.newuser.do_something(user1)oritshouldbeuser_controller=UserController.newuser_controller.do_something(user1,user2)我
什么是ruby的rack或python的Java的wsgi?还有一个路由库。 最佳答案 来自Python标准PEP333:Bycontrast,althoughJavahasjustasmanywebapplicationframeworksavailable,Java's"servlet"APImakesitpossibleforapplicationswrittenwithanyJavawebapplicationframeworktoruninanywebserverthatsupportstheservletAPI.ht
这篇文章是继上一篇文章“Observability:从零开始创建Java微服务并监控它(一)”的续篇。在上一篇文章中,我们讲述了如何创建一个Javaweb应用,并使用Filebeat来收集应用所生成的日志。在今天的文章中,我来详述如何收集应用的指标,使用APM来监控应用并监督web服务的在线情况。源码可以在地址 https://github.com/liu-xiao-guo/java_observability 进行下载。摄入指标指标被视为可以随时更改的时间点值。当前请求的数量可以改变任何毫秒。你可能有1000个请求的峰值,然后一切都回到一个请求。这也意味着这些指标可能不准确,你还想提取最小/
HashMap中为什么引入红黑树,而不是AVL树呢1.概述开始学习这个知识点之前我们需要知道,在JDK1.8以及之前,针对HashMap有什么不同。JDK1.7的时候,HashMap的底层实现是数组+链表JDK1.8的时候,HashMap的底层实现是数组+链表+红黑树我们要思考一个问题,为什么要从链表转为红黑树呢。首先先让我们了解下链表有什么不好???2.链表上述的截图其实就是链表的结构,我们来看下链表的增删改查的时间复杂度增:因为链表不是线性结构,所以每次添加的时候,只需要移动一个节点,所以可以理解为复杂度是N(1)删:算法时间复杂度跟增保持一致查:既然是非线性结构,所以查询某一个节点的时候