【发布时间】：2017-03-10 18:05:02
【问题描述】：
我使用的是 Active MQ 5.10 版本，我已经配置了 wso2esb 用于消息处理。
大约 7-10 天后，Active MQ 抛出 tcp 连接失败异常，因为 ESB 没有接收到成功的 tcp 连接，所以无法在队列中提交消息。
在这种情况下，我重新启动服务器，它会再次运行 7-10 天，并且重复同样的事情。
我的问题是
活动 MQ 停止提供成功的 tcp 连接的确切原因可能是什么？
为什么重启服务器后又回到正常状态..？
有没有最好的解决方案来解决这个问题..
activemq.xml文件中的内存配置
<systemUsage>
<systemUsage sendFailIfNoSpace="true">
<memoryUsage>
<memoryUsage limit="1430 mb"/>
</memoryUsage>
<storeUsage>
<storeUsage limit="300 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="100 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>
这是将消息放入队列的代理服务。在代理服务中，它将检查用户身份验证是否为真，用户可以将消息放入队列中，这里我使用一个类调解器，它与活动 MQ 连接并放置消息
<proxy xmlns="http://ws.apache.org/ns/synapse"
name="JmsStore2.0"
transports="https http"
startOnLoad="true"
trace="disable"
statistics="enable">
<description/>
<target>
<inSequence onError="fault">
<property name="messageType" value="application/json" scope="axis2"/>
<property name="FORCE_ERROR_ON_SOAP_FAULT" value="true"/>
<property name="jmsuri" value="tcp://0.0.0.0:61616"/>
<property name="jmsqueue" expression="get-property('transport', 'jmsqueue')"/>
<property name="readingspayload" expression="$body" type="OM"/>
<property name="username" expression="get-property('transport', 'username')"/>
<property name="password" expression="get-property('transport', 'password')"/>
<property name="PartyBranchID"
expression="//FieldValue/text()"
scope="default"
type="STRING"/>
<property name="Body" expression="$body" scope="default" type="STRING"/>
<property name="usercode"
expression="fn:substring-before(get-property('username'),'|')"
scope="default"
type="STRING"/>
<property name="clientid"
expression="fn:substring-after(get-property('username'),'|')"
scope="default"
type="STRING"/>
<property name="requestMsgId"
expression="get-property('MessageID')"
scope="default"
type="STRING"/>
<property name="client_ip_address"
expression="get-property('axis2','REMOTE_ADDR')"
scope="default"
type="STRING"/>
<payloadFactory media-type="xml">
<format>
<send xmlns="">
<username>$1</username>
<password>$2</password>
</send>
</format>
<args>
<arg evaluator="xml" expression="get-property('username')"/>
<arg evaluator="xml" expression="get-property('password')"/>
</args>
</payloadFactory>
<send receive="JmsStore_Seq">
<endpoint>
<address uri="http://localhost:8282/services/Login2.0" format="soap11">
<suspendOnFailure>
<errorCodes>101500,101501,101506,101507,101508,101503,50000</errorCodes>
<initialDuration>30</initialDuration>
<progressionFactor>1.0</progressionFactor>
<maximumDuration>300</maximumDuration>
</suspendOnFailure>
</address>
</endpoint>
</send>
</inSequence>
<outSequence onError="fault">
<send/>
</outSequence>
</target>
</proxy>
顺序：
<sequence xmlns="http://ws.apache.org/ns/synapse"
name="JmsStore_Seq"
trace="disable">
<property name="FORCE_ERROR_ON_SOAP_FAULT" value="true"/>
<property xmlns:ns="http://org.apache.synapse/xsd"
name="Authentication"
expression="//Authentication/text()"/>
<property xmlns:ns="http://org.apache.synapse/xsd"
name="UserId"
expression="//UserId/text()"
scope="default"
type="STRING"/>
<property xmlns:ns="http://org.apache.synapse/xsd"
name="WorkOUid"
expression="//WorkOUid/text()"/>
<property xmlns:ns="http://org.apache.synapse/xsd"
name="WorkPartyBranchId"
expression="//WorkPartyBranchId/text()"/>
<filter xmlns:ns="http://org.apache.synapse/xsd"
xpath="get-property('Authentication')=''">
<then>
<payloadFactory media-type="xml">
<format>
<ResponseJSON xmlns="">
<Exception>Service trying to connect inactive service</Exception>
<Status>101503</Status>
</ResponseJSON>
</format>
<args/>
</payloadFactory>
<property name="messageType" value="application/json" scope="axis2"/>
<property name="HTTP_METHOD" value="POST" scope="axis2" type="STRING"/>
<property name="RESPONSE" value="true" scope="default" type="STRING"/>
<property name="NO_ENTITY_BODY" scope="axis2" action="remove"/>
<send/>
</then>
<else>
<filter xpath="get-property('Authentication')='false'">
<then>
<payloadFactory media-type="xml">
<format>
<ResponseJSON xmlns="">
<Exception>Authentication Failed</Exception>
<Status>401</Status>
</ResponseJSON>
</format>
<args/>
</payloadFactory>
<property name="messageType" value="application/json" scope="axis2"/>
<property name="HTTP_METHOD" value="POST" scope="axis2" type="STRING"/>
<property name="RESPONSE" value="true" scope="default" type="STRING"/>
<property name="NO_ENTITY_BODY" scope="axis2" action="remove"/>
<send/>
</then>
<else>
<property name="jmspayload"
expression="get-property('readingspayload')"
type="OM"/>
<property name="ResponseJSON" expression="$body/ResponseJSON" type="OM"/>
<property name="jmsuri" expression="get-property('jmsuri')"/>
<property name="jmsqueue" expression="get-property('jmsqueue')"/>
<payloadFactory media-type="xml">
<format>
<PLData>
<JMpayload>$1</JMpayload>
<AuthData>$2</AuthData>
<LogData>
<usercode>$3</usercode>
<clientid>$4</clientid>
<requestMsgId>$5</requestMsgId>
</LogData>
</PLData>
</format>
<args>
<arg evaluator="xml" expression="get-property('jmspayload')"/>
<arg evaluator="xml" expression="get-property('ResponseJSON')"/>
<arg evaluator="xml" expression="get-property('usercode')"/>
<arg evaluator="xml" expression="get-property('clientid')"/>
<arg evaluator="xml" expression="get-property('requestMsgId')"/>
</args>
</payloadFactory>
<class name="in.youtility.esb.custommediators.JMSStoreMediator"/>
<payloadFactory media-type="xml">
<format>
<ResponseJSON xmlns="">
<Body>
<Datalist>
<Data>Successfully stored</Data>
</Datalist>
</Body>
<Status>200</Status>
</ResponseJSON>
</format>
<args/>
</payloadFactory>
<property name="messageType" value="application/json" scope="axis2"/>
<header name="To" action="remove"/>
<property name="NO_ENTITY_BODY" scope="axis2" action="remove"/>
<property name="RESPONSE" value="true"/>
<send/>
</else>
</filter>
</else>
</filter>
<description/>
</sequence>
类中介：
public class JMSStoreMediator extends AbstractMediator implements
托管生命周期{
Connection connection;
public boolean mediate(MessageContext msgCtx) {
try {
boolean topic=false;
String jmsuri=""+msgCtx.getProperty("jmsuri");
String t=""+msgCtx.getProperty("topic");
if(t.isEmpty()){
topic=false;
}
else {
topic=Boolean.valueOf(t);
}
ConnectionFactory factory= new ActiveMQConnectionFactory(jmsuri);
connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination=null;
if(!topic)destination= session.createQueue(""+msgCtx.getProperty("jmsqueue"));
else destination= session.createTopic(""+msgCtx.getProperty("jmsqueue"));
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
String xml = ""+msgCtx.getEnvelope().getBody().toStringWithConsume();
if(topic){
JSONObject obj=XML.toJSONObject(xml);
JSONObject ar=obj.getJSONObject("soapenv:Body");
ar.remove("xmlns:soapenv");
xml=ar.toString();
}
TextMessage message = session.createTextMessage(xml);
producer.send(message);
} catch (Exception e) {
log.info("LogLocation = "+getClass().getName()+",Error in storing message in JMS stacktrace is :"+e.toString());
((Axis2MessageContext) msgCtx).setProperty(NhttpConstants.HTTP_SC, 500);
handleException("Error while storing in the message store", msgCtx);
}
finally {
try {
connection.close();
} catch (JMSException e) {
log.info("LogLocation = "+getClass().getName()+",Error in closing JMS connection stacktrace is :"+e.toString());
}
}
log.info("LogLocation = "+getClass().getName()+",ProxyName = "+msgCtx.getProperty("proxy.name")+
",Usercode = "+msgCtx.getProperty("usercode")+",Clientid = "+msgCtx.getProperty("clientid")+
",requestMsgId = "+msgCtx.getProperty("requestMsgId")+",Position = END");
return true;
}
【问题讨论】：
-
ActiveMQ 中的消息数量是否每天都在增长？您如何使用带有 JMS 代理服务、MessageStore 的 AMQ？
-
嗨@Jean-Michel 感谢您的回复，我正在使用 ESB 代理服务，它将在队列中放置一条消息，而其他 JMS 代理服务将侦听此队列并处理它。消息经常进入队列并得到处理，直到 Active MQ 成功提供与 ESB 的 tcp 连接。
-
我发布了第一个答案，但是告诉我如果开始失败时 ActiveMQ 中的消息数量是否在增长（队列中有数千条消息等待）
-
出现此问题时队列中没有等待消息。收到的所有消息都会立即处理。
标签： tcp jms activemq wso2esb wso2carbon